1 # -*- coding: utf-8 -*-
4 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
19 from http
import HTTPStatus
20 from itertools
import product
22 from random
import choice
as random_choice
23 from threading
import Lock
25 from traceback
import format_exc
as traceback_format_exc
26 from typing
import Any
, Dict
, List
, Optional
, Tuple
, Type
27 from uuid
import uuid4
29 from cryptography
.hazmat
.backends
import default_backend
as crypto_default_backend
30 from cryptography
.hazmat
.primitives
import serialization
as crypto_serialization
31 from cryptography
.hazmat
.primitives
.asymmetric
import rsa
40 from osm_common
import (
47 version
as common_version
,
49 from osm_common
.dbbase
import DbBase
, DbException
50 from osm_common
.fsbase
import FsBase
, FsException
51 from osm_common
.msgbase
import MsgException
52 from osm_ng_ro
.ns_thread
import deep_get
, NsWorker
, NsWorkerException
53 from osm_ng_ro
.validation
import deploy_schema
, validate_input
56 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
57 min_common_version
= "0.1.16"
60 class NsException(Exception):
61 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
62 self
.http_code
= http_code
63 super(Exception, self
).__init
__(message
)
68 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
69 will provide a random one
72 # Try getting docker id. If fails, get pid
74 with
open("/proc/self/cgroup", "r") as f
:
75 text_id_
= f
.readline()
76 _
, _
, text_id
= text_id_
.rpartition("/")
77 text_id
= text_id
.replace("\n", "")[:12]
81 except Exception as error
:
82 logging
.exception(f
"{error} occured while getting process id")
85 return "".join(random_choice("0123456789abcdef") for _
in range(12))
89 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
92 for point
in v
.split("."):
93 filled
.append(point
.zfill(8))
104 # self.operations = None
106 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
107 # If done now it will not be linked to parent not getting its handler and level
109 self
.write_lock
= None
110 self
.vims_assigned
= {}
114 self
.process_params_function_map
= {
115 "net": Ns
._process
_net
_params
,
116 "image": Ns
._process
_image
_params
,
117 "flavor": Ns
._process
_flavor
_params
,
118 "vdu": Ns
._process
_vdu
_params
,
119 "affinity-or-anti-affinity-group": Ns
._process
_affinity
_group
_params
,
126 "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
129 def init_db(self
, target_version
):
132 def start(self
, config
):
134 Connect to database, filesystem storage, and messaging
135 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
136 :param config: Configuration of db, storage, etc
140 self
.config
["process_id"] = get_process_id() # used for HA identity
141 self
.logger
= logging
.getLogger("ro.ns")
143 # check right version of common
144 if versiontuple(common_version
) < versiontuple(min_common_version
):
146 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
147 common_version
, min_common_version
153 if config
["database"]["driver"] == "mongo":
154 self
.db
= dbmongo
.DbMongo()
155 self
.db
.db_connect(config
["database"])
156 elif config
["database"]["driver"] == "memory":
157 self
.db
= dbmemory
.DbMemory()
158 self
.db
.db_connect(config
["database"])
161 "Invalid configuration param '{}' at '[database]':'driver'".format(
162 config
["database"]["driver"]
167 if config
["storage"]["driver"] == "local":
168 self
.fs
= fslocal
.FsLocal()
169 self
.fs
.fs_connect(config
["storage"])
170 elif config
["storage"]["driver"] == "mongo":
171 self
.fs
= fsmongo
.FsMongo()
172 self
.fs
.fs_connect(config
["storage"])
173 elif config
["storage"]["driver"] is None:
177 "Invalid configuration param '{}' at '[storage]':'driver'".format(
178 config
["storage"]["driver"]
183 if config
["message"]["driver"] == "local":
184 self
.msg
= msglocal
.MsgLocal()
185 self
.msg
.connect(config
["message"])
186 elif config
["message"]["driver"] == "kafka":
187 self
.msg
= msgkafka
.MsgKafka()
188 self
.msg
.connect(config
["message"])
191 "Invalid configuration param '{}' at '[message]':'driver'".format(
192 config
["message"]["driver"]
196 # TODO load workers to deal with exising database tasks
198 self
.write_lock
= Lock()
199 except (DbException
, FsException
, MsgException
) as e
:
200 raise NsException(str(e
), http_code
=e
.http_code
)
202 def get_assigned_vims(self
):
203 return list(self
.vims_assigned
.keys())
208 self
.db
.db_disconnect()
211 self
.fs
.fs_disconnect()
214 self
.msg
.disconnect()
216 self
.write_lock
= None
217 except (DbException
, FsException
, MsgException
) as e
:
218 raise NsException(str(e
), http_code
=e
.http_code
)
220 for worker
in self
.workers
:
221 worker
.insert_task(("terminate",))
223 def _create_worker(self
):
225 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
226 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
227 return the index of the assigned worker thread. Worker threads are storead at self.workers
229 # Look for a thread in idle status
233 for i
in range(len(self
.workers
))
234 if self
.workers
[i
] and self
.workers
[i
].idle
239 if worker_id
is not None:
240 # unset idle status to avoid race conditions
241 self
.workers
[worker_id
].idle
= False
243 worker_id
= len(self
.workers
)
245 if worker_id
< self
.config
["global"]["server.ns_threads"]:
246 # create a new worker
248 NsWorker(worker_id
, self
.config
, self
.plugins
, self
.db
)
250 self
.workers
[worker_id
].start()
252 # reached maximum number of threads, assign VIM to an existing one
253 worker_id
= self
.next_worker
254 self
.next_worker
= (self
.next_worker
+ 1) % self
.config
["global"][
260 def assign_vim(self
, target_id
):
261 with self
.write_lock
:
262 return self
._assign
_vim
(target_id
)
264 def _assign_vim(self
, target_id
):
265 if target_id
not in self
.vims_assigned
:
266 worker_id
= self
.vims_assigned
[target_id
] = self
._create
_worker
()
267 self
.workers
[worker_id
].insert_task(("load_vim", target_id
))
269 def reload_vim(self
, target_id
):
270 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
271 # this is because database VIM information is cached for threads working with SDN
272 with self
.write_lock
:
273 for worker
in self
.workers
:
274 if worker
and not worker
.idle
:
275 worker
.insert_task(("reload_vim", target_id
))
277 def unload_vim(self
, target_id
):
278 with self
.write_lock
:
279 return self
._unload
_vim
(target_id
)
281 def _unload_vim(self
, target_id
):
282 if target_id
in self
.vims_assigned
:
283 worker_id
= self
.vims_assigned
[target_id
]
284 self
.workers
[worker_id
].insert_task(("unload_vim", target_id
))
285 del self
.vims_assigned
[target_id
]
287 def check_vim(self
, target_id
):
288 with self
.write_lock
:
289 if target_id
in self
.vims_assigned
:
290 worker_id
= self
.vims_assigned
[target_id
]
292 worker_id
= self
._create
_worker
()
294 worker
= self
.workers
[worker_id
]
295 worker
.insert_task(("check_vim", target_id
))
297 def unload_unused_vims(self
):
298 with self
.write_lock
:
301 for target_id
in self
.vims_assigned
:
302 if not self
.db
.get_one(
305 "target_id": target_id
,
306 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
310 vims_to_unload
.append(target_id
)
312 for target_id
in vims_to_unload
:
313 self
._unload
_vim
(target_id
)
321 """This method reads cloud init from a file.
323 Note: Not used as cloud init content is provided in the http body.
326 db (Type[DbBase]): [description]
327 fs (Type[FsBase]): [description]
328 location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
331 NsException: [description]
332 NsException: [description]
337 vnfd_id
, _
, other
= location
.partition(":")
338 _type
, _
, name
= other
.partition(":")
339 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
342 base_folder
= vnfd
["_admin"]["storage"]
343 cloud_init_file
= "{}/{}/cloud_init/{}".format(
344 base_folder
["folder"], base_folder
["pkg-dir"], name
349 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
354 with fs
.file_open(cloud_init_file
, "r") as ci_file
:
355 cloud_init_content
= ci_file
.read()
357 cloud_init_content
= vnfd
["vdu"][int(name
)]["cloud-init"]
359 raise NsException("Mismatch descriptor for cloud init: {}".format(location
))
361 return cloud_init_content
365 cloud_init_content
: str,
366 params
: Dict
[str, Any
],
369 """Function that processes the cloud init to replace Jinja2 encoded parameters.
372 cloud_init_content (str): [description]
373 params (Dict[str, Any]): [description]
374 context (str): [description]
377 NsException: [description]
378 NsException: [description]
385 undefined
=StrictUndefined
,
386 autoescape
=select_autoescape(default_for_string
=True, default
=True),
388 template
= env
.from_string(cloud_init_content
)
390 return template
.render(params
or {})
391 except UndefinedError
as e
:
393 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
394 "inside the 'additionalParamsForVnf' block".format(e
, context
)
396 except (TemplateError
, TemplateNotFound
) as e
:
398 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
403 def _create_db_ro_nsrs(self
, nsr_id
, now
):
405 key
= rsa
.generate_private_key(
406 backend
=crypto_default_backend(), public_exponent
=65537, key_size
=2048
408 private_key
= key
.private_bytes(
409 crypto_serialization
.Encoding
.PEM
,
410 crypto_serialization
.PrivateFormat
.PKCS8
,
411 crypto_serialization
.NoEncryption(),
413 public_key
= key
.public_key().public_bytes(
414 crypto_serialization
.Encoding
.OpenSSH
,
415 crypto_serialization
.PublicFormat
.OpenSSH
,
417 private_key
= private_key
.decode("utf8")
418 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
419 i
= private_key
.find("\n")
420 private_key
= "-----BEGIN RSA PRIVATE KEY-----" + private_key
[i
:]
421 public_key
= public_key
.decode("utf8")
422 except Exception as e
:
423 raise NsException("Cannot create ssh-keys: {}".format(e
))
425 schema_version
= "1.1"
426 private_key_encrypted
= self
.db
.encrypt(
427 private_key
, schema_version
=schema_version
, salt
=nsr_id
434 "schema_version": schema_version
,
436 "public_key": public_key
,
437 "private_key": private_key_encrypted
,
440 self
.db
.create("ro_nsrs", db_content
)
446 deployment_info
: Dict
[str, Any
],
451 target_record_id
: str,
452 extra_dict
: Dict
[str, Any
] = None,
454 """Function to create task dict from deployment information.
457 deployment_info (Dict[str, Any]): [description]
458 target_id (str): [description]
459 item (str): [description]
460 action (str): [description]
461 target_record (str): [description]
462 target_record_id (str): [description]
463 extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
466 Dict[str, Any]: [description]
469 "target_id": target_id
, # it will be removed before pushing at database
470 "action_id": deployment_info
.get("action_id"),
471 "nsr_id": deployment_info
.get("nsr_id"),
472 "task_id": f
"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
473 "status": "SCHEDULED",
476 "target_record": target_record
,
477 "target_record_id": target_record_id
,
481 task
.update(extra_dict
) # params, find_params, depends_on
483 deployment_info
["task_index"] = deployment_info
.get("task_index", 0) + 1
490 task
: Dict
[str, Any
],
492 """Function to create an RO task from task information.
495 target_id (str): [description]
496 task (Dict[str, Any]): [description]
499 Dict[str, Any]: [description]
503 _id
= task
.get("task_id")
508 "target_id": target_id
,
511 "created_items": None,
528 def _process_image_params(
529 target_image
: Dict
[str, Any
],
530 indata
: Dict
[str, Any
],
531 vim_info
: Dict
[str, Any
],
532 target_record_id
: str,
533 **kwargs
: Dict
[str, Any
],
535 """Function to process VDU image parameters.
538 target_image (Dict[str, Any]): [description]
539 indata (Dict[str, Any]): [description]
540 vim_info (Dict[str, Any]): [description]
541 target_record_id (str): [description]
544 Dict[str, Any]: [description]
548 if target_image
.get("image"):
549 find_params
["filter_dict"] = {"name": target_image
.get("image")}
551 if target_image
.get("vim_image_id"):
552 find_params
["filter_dict"] = {"id": target_image
.get("vim_image_id")}
554 if target_image
.get("image_checksum"):
555 find_params
["filter_dict"] = {
556 "checksum": target_image
.get("image_checksum")
559 return {"find_params": find_params
}
562 def _get_resource_allocation_params(
563 quota_descriptor
: Dict
[str, Any
],
565 """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
569 quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
572 Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
576 if quota_descriptor
.get("limit"):
577 quota
["limit"] = int(quota_descriptor
["limit"])
579 if quota_descriptor
.get("reserve"):
580 quota
["reserve"] = int(quota_descriptor
["reserve"])
582 if quota_descriptor
.get("shares"):
583 quota
["shares"] = int(quota_descriptor
["shares"])
588 def _process_guest_epa_quota_params(
589 guest_epa_quota
: Dict
[str, Any
],
592 """Function to extract the guest epa quota parameters.
595 guest_epa_quota (Dict[str, Any]): [description]
596 epa_vcpu_set (bool): [description]
599 Dict[str, Any]: [description]
603 if guest_epa_quota
.get("cpu-quota") and not epa_vcpu_set
:
604 cpuquota
= Ns
._get
_resource
_allocation
_params
(
605 guest_epa_quota
.get("cpu-quota")
609 result
["cpu-quota"] = cpuquota
611 if guest_epa_quota
.get("mem-quota"):
612 vduquota
= Ns
._get
_resource
_allocation
_params
(
613 guest_epa_quota
.get("mem-quota")
617 result
["mem-quota"] = vduquota
619 if guest_epa_quota
.get("disk-io-quota"):
620 diskioquota
= Ns
._get
_resource
_allocation
_params
(
621 guest_epa_quota
.get("disk-io-quota")
625 result
["disk-io-quota"] = diskioquota
627 if guest_epa_quota
.get("vif-quota"):
628 vifquota
= Ns
._get
_resource
_allocation
_params
(
629 guest_epa_quota
.get("vif-quota")
633 result
["vif-quota"] = vifquota
638 def _process_guest_epa_numa_params(
639 guest_epa_quota
: Dict
[str, Any
],
640 ) -> Tuple
[Dict
[str, Any
], bool]:
644 guest_epa_quota (Dict[str, Any]): [description]
647 Tuple[Dict[str, Any], bool]: [description]
653 if guest_epa_quota
.get("numa-node-policy"):
654 numa_node_policy
= guest_epa_quota
.get("numa-node-policy")
656 if numa_node_policy
.get("node"):
657 for numa_node
in numa_node_policy
["node"]:
659 if numa_node
.get("id"):
660 numa
["id"] = int(numa_node
["id"])
662 if numa_node
.get("vcpu"):
663 for vcpu
in numa_node
.get("vcpu"):
664 vcpu_id
= int(vcpu
.get("id"))
665 vcpu_list
.append(vcpu_id
)
666 numa
["vcpu"] = vcpu_list
668 if numa_node
.get("num-cores"):
669 numa
["cores"] = numa_node
["num-cores"]
672 paired_threads
= numa_node
.get("paired-threads", {})
673 if paired_threads
.get("num-paired-threads"):
674 numa
["paired_threads"] = int(
675 numa_node
["paired-threads"]["num-paired-threads"]
679 if paired_threads
.get("paired-thread-ids"):
680 numa
["paired-threads-id"] = []
682 for pair
in paired_threads
["paired-thread-ids"]:
683 numa
["paired-threads-id"].append(
685 str(pair
["thread-a"]),
686 str(pair
["thread-b"]),
690 if numa_node
.get("num-threads"):
691 numa
["threads"] = int(numa_node
["num-threads"])
694 if numa_node
.get("memory-mb"):
695 numa
["memory"] = max(int(int(numa_node
["memory-mb"]) / 1024), 1)
697 numa_list
.append(numa
)
700 return numa_list
, epa_vcpu_set
703 def _process_guest_epa_cpu_pinning_params(
704 guest_epa_quota
: Dict
[str, Any
],
707 ) -> Tuple
[Dict
[str, Any
], bool]:
711 guest_epa_quota (Dict[str, Any]): [description]
712 vcpu_count (int): [description]
713 epa_vcpu_set (bool): [description]
716 Tuple[Dict[str, Any], bool]: [description]
719 local_epa_vcpu_set
= epa_vcpu_set
722 guest_epa_quota
.get("cpu-pinning-policy") == "DEDICATED"
725 # Pinning policy "REQUIRE" uses threads as host should support SMT architecture
726 # Pinning policy "ISOLATE" uses cores as host should not support SMT architecture
727 # Pinning policy "PREFER" uses threads in case host supports SMT architecture
730 if guest_epa_quota
.get("cpu-thread-pinning-policy") == "ISOLATE"
732 ] = max(vcpu_count
, 1)
733 local_epa_vcpu_set
= True
735 return numa
, local_epa_vcpu_set
738 def _process_epa_params(
739 target_flavor
: Dict
[str, Any
],
744 target_flavor (Dict[str, Any]): [description]
747 Dict[str, Any]: [description]
753 if target_flavor
.get("guest-epa"):
754 guest_epa
= target_flavor
["guest-epa"]
756 numa_list
, epa_vcpu_set
= Ns
._process
_guest
_epa
_numa
_params
(
757 guest_epa_quota
=guest_epa
760 if guest_epa
.get("mempage-size"):
761 extended
["mempage-size"] = guest_epa
.get("mempage-size")
763 if guest_epa
.get("cpu-pinning-policy"):
764 extended
["cpu-pinning-policy"] = guest_epa
.get("cpu-pinning-policy")
766 if guest_epa
.get("cpu-thread-pinning-policy"):
767 extended
["cpu-thread-pinning-policy"] = guest_epa
.get(
768 "cpu-thread-pinning-policy"
771 if guest_epa
.get("numa-node-policy"):
772 if guest_epa
.get("numa-node-policy").get("mem-policy"):
773 extended
["mem-policy"] = guest_epa
.get("numa-node-policy").get(
777 tmp_numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
778 guest_epa_quota
=guest_epa
,
779 vcpu_count
=int(target_flavor
.get("vcpu-count", 1)),
780 epa_vcpu_set
=epa_vcpu_set
,
782 for numa
in numa_list
:
783 numa
.update(tmp_numa
)
786 Ns
._process
_guest
_epa
_quota
_params
(
787 guest_epa_quota
=guest_epa
,
788 epa_vcpu_set
=epa_vcpu_set
,
793 extended
["numas"] = numa_list
798 def _process_flavor_params(
799 target_flavor
: Dict
[str, Any
],
800 indata
: Dict
[str, Any
],
801 vim_info
: Dict
[str, Any
],
802 target_record_id
: str,
803 **kwargs
: Dict
[str, Any
],
808 target_flavor (Dict[str, Any]): [description]
809 indata (Dict[str, Any]): [description]
810 vim_info (Dict[str, Any]): [description]
811 target_record_id (str): [description]
814 Dict[str, Any]: [description]
816 db
= kwargs
.get("db")
820 "disk": int(target_flavor
["storage-gb"]),
821 "ram": int(target_flavor
["memory-mb"]),
822 "vcpus": int(target_flavor
["vcpu-count"]),
825 for vnf
in indata
.get("vnf", []):
826 for vdur
in vnf
.get("vdur", []):
827 if vdur
.get("ns-flavor-id") == target_flavor
.get("id"):
830 if db
and isinstance(indata
.get("vnf"), list):
831 vnfd_id
= indata
.get("vnf")[0].get("vnfd-id")
832 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
833 # check if there is persistent root disk
834 for vdu
in vnfd
.get("vdu", ()):
835 if vdu
["name"] == target_vdur
.get("vdu-name"):
836 for vsd
in vnfd
.get("virtual-storage-desc", ()):
837 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
840 root_disk
.get("type-of-storage")
841 == "persistent-storage:persistent-storage"
843 flavor_data
["disk"] = 0
845 for storage
in target_vdur
.get("virtual-storages", []):
847 storage
.get("type-of-storage")
848 == "etsi-nfv-descriptors:ephemeral-storage"
850 flavor_data
["ephemeral"] = int(storage
.get("size-of-storage", 0))
851 elif storage
.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
852 flavor_data
["swap"] = int(storage
.get("size-of-storage", 0))
854 extended
= Ns
._process
_epa
_params
(target_flavor
)
856 flavor_data
["extended"] = extended
858 extra_dict
= {"find_params": {"flavor_data": flavor_data
}}
859 flavor_data_name
= flavor_data
.copy()
860 flavor_data_name
["name"] = target_flavor
["name"]
861 extra_dict
["params"] = {"flavor_data": flavor_data_name
}
866 def _ip_profile_to_ro(
867 ip_profile
: Dict
[str, Any
],
872 ip_profile (Dict[str, Any]): [description]
875 Dict[str, Any]: [description]
882 if "v4" in ip_profile
.get("ip-version", "ipv4")
884 "subnet_address": ip_profile
.get("subnet-address"),
885 "gateway_address": ip_profile
.get("gateway-address"),
886 "dhcp_enabled": ip_profile
.get("dhcp-params", {}).get("enabled", False),
887 "dhcp_start_address": ip_profile
.get("dhcp-params", {}).get(
888 "start-address", None
890 "dhcp_count": ip_profile
.get("dhcp-params", {}).get("count", None),
893 if ip_profile
.get("dns-server"):
894 ro_ip_profile
["dns_address"] = ";".join(
895 [v
["address"] for v
in ip_profile
["dns-server"] if v
.get("address")]
898 if ip_profile
.get("security-group"):
899 ro_ip_profile
["security_group"] = ip_profile
["security-group"]
904 def _process_net_params(
905 target_vld
: Dict
[str, Any
],
906 indata
: Dict
[str, Any
],
907 vim_info
: Dict
[str, Any
],
908 target_record_id
: str,
909 **kwargs
: Dict
[str, Any
],
911 """Function to process network parameters.
914 target_vld (Dict[str, Any]): [description]
915 indata (Dict[str, Any]): [description]
916 vim_info (Dict[str, Any]): [description]
917 target_record_id (str): [description]
920 Dict[str, Any]: [description]
924 if vim_info
.get("sdn"):
925 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
926 # ns_preffix = "nsrs:{}".format(nsr_id)
927 # remove the ending ".sdn
928 vld_target_record_id
, _
, _
= target_record_id
.rpartition(".")
929 extra_dict
["params"] = {
931 for k
in ("sdn-ports", "target_vim", "vlds", "type")
935 # TODO needed to add target_id in the dependency.
936 if vim_info
.get("target_vim"):
937 extra_dict
["depends_on"] = [
938 f
"{vim_info.get('target_vim')} {vld_target_record_id}"
943 if vim_info
.get("vim_network_name"):
944 extra_dict
["find_params"] = {
946 "name": vim_info
.get("vim_network_name"),
949 elif vim_info
.get("vim_network_id"):
950 extra_dict
["find_params"] = {
952 "id": vim_info
.get("vim_network_id"),
955 elif target_vld
.get("mgmt-network") and not vim_info
.get("provider_network"):
956 extra_dict
["find_params"] = {
958 "name": target_vld
["id"],
962 extra_dict
["params"] = {
964 f
"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
966 "ip_profile": Ns
._ip
_profile
_to
_ro
(vim_info
.get("ip_profile")),
967 "provider_network_profile": vim_info
.get("provider_network"),
970 if not target_vld
.get("underlay"):
971 extra_dict
["params"]["net_type"] = "bridge"
973 extra_dict
["params"]["net_type"] = (
974 "ptp" if target_vld
.get("type") == "ELINE" else "data"
980 def find_persistent_root_volumes(
983 vdu_instantiation_volumes_list
: list,
986 """Find the persistent root volumes and add them to the disk_list
987 by parsing the instantiation parameters.
990 vnfd (dict): VNF descriptor
991 target_vdu (dict): processed VDU
992 vdu_instantiation_volumes_list (list): instantiation parameters for the each VDU as a list
993 disk_list (list): to be filled up
996 persistent_root_disk (dict): Details of persistent root disk
999 persistent_root_disk
= {}
1000 # There can be only one root disk, when we find it, it will return the result
1002 for vdu
, vsd
in product(
1003 vnfd
.get("vdu", ()), vnfd
.get("virtual-storage-desc", ())
1006 vdu
["name"] == target_vdu
["vdu-name"]
1007 and vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]
1011 root_disk
.get("type-of-storage")
1012 == "persistent-storage:persistent-storage"
1014 for vdu_volume
in vdu_instantiation_volumes_list
:
1016 vdu_volume
["vim-volume-id"]
1017 and root_disk
["id"] == vdu_volume
["name"]
1019 persistent_root_disk
[vsd
["id"]] = {
1020 "vim_volume_id": vdu_volume
["vim-volume-id"],
1021 "image_id": vdu
.get("sw-image-desc"),
1024 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1026 return persistent_root_disk
1029 if root_disk
.get("size-of-storage"):
1030 persistent_root_disk
[vsd
["id"]] = {
1031 "image_id": vdu
.get("sw-image-desc"),
1032 "size": root_disk
.get("size-of-storage"),
1035 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1037 return persistent_root_disk
1040 def find_persistent_volumes(
1041 persistent_root_disk
: dict,
1043 vdu_instantiation_volumes_list
: list,
1046 """Find the ordinary persistent volumes and add them to the disk_list
1047 by parsing the instantiation parameters.
1050 persistent_root_disk: persistent root disk dictionary
1051 target_vdu: processed VDU
1052 vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list
1053 disk_list: to be filled up
1056 # Find the ordinary volumes which are not added to the persistent_root_disk
1057 persistent_disk
= {}
1058 for disk
in target_vdu
.get("virtual-storages", {}):
1060 disk
.get("type-of-storage") == "persistent-storage:persistent-storage"
1061 and disk
["id"] not in persistent_root_disk
.keys()
1063 for vdu_volume
in vdu_instantiation_volumes_list
:
1064 if vdu_volume
["vim-volume-id"] and disk
["id"] == vdu_volume
["name"]:
1065 persistent_disk
[disk
["id"]] = {
1066 "vim_volume_id": vdu_volume
["vim-volume-id"],
1068 disk_list
.append(persistent_disk
[disk
["id"]])
1071 if disk
["id"] not in persistent_disk
.keys():
1072 persistent_disk
[disk
["id"]] = {
1073 "size": disk
.get("size-of-storage"),
1075 disk_list
.append(persistent_disk
[disk
["id"]])
1078 def _sort_vdu_interfaces(target_vdu
: dict) -> None:
1079 """Sort the interfaces according to position number.
1082 target_vdu (dict): Details of VDU to be created
1085 # If the position info is provided for all the interfaces, it will be sorted
1086 # according to position number ascendingly.
1087 sorted_interfaces
= sorted(
1088 target_vdu
["interfaces"],
1089 key
=lambda x
: (x
.get("position") is None, x
.get("position")),
1091 target_vdu
["interfaces"] = sorted_interfaces
1094 def _partially_locate_vdu_interfaces(target_vdu
: dict) -> None:
1095 """Only place the interfaces which has specific position.
1098 target_vdu (dict): Details of VDU to be created
1101 # If the position info is provided for some interfaces but not all of them, the interfaces
1102 # which has specific position numbers will be placed and others' positions will not be taken care.
1104 i
.get("position") + 1
1105 for i
in target_vdu
["interfaces"]
1106 if i
.get("position") is not None
1108 n
= len(target_vdu
["interfaces"])
1109 sorted_interfaces
= [-1] * n
1113 if target_vdu
["interfaces"][k
].get("position") is not None:
1114 if any(i
.get("position") == 0 for i
in target_vdu
["interfaces"]):
1115 idx
= target_vdu
["interfaces"][k
]["position"] + 1
1117 idx
= target_vdu
["interfaces"][k
]["position"]
1118 sorted_interfaces
[idx
- 1] = target_vdu
["interfaces"][k
]
1122 if target_vdu
["interfaces"][m
].get("position") is None:
1123 idy
= sorted_interfaces
.index(-1)
1124 sorted_interfaces
[idy
] = target_vdu
["interfaces"][m
]
1127 target_vdu
["interfaces"] = sorted_interfaces
1130 def _prepare_vdu_cloud_init(
1131 target_vdu
: dict, vdu2cloud_init
: dict, db
: object, fs
: object
1133 """Fill cloud_config dict with cloud init details.
1136 target_vdu (dict): Details of VDU to be created
1137 vdu2cloud_init (dict): Cloud init dict
1138 db (object): DB object
1139 fs (object): FS object
1142 cloud_config (dict): Cloud config details of VDU
1148 if target_vdu
.get("cloud-init"):
1149 if target_vdu
["cloud-init"] not in vdu2cloud_init
:
1150 vdu2cloud_init
[target_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1153 location
=target_vdu
["cloud-init"],
1156 cloud_content_
= vdu2cloud_init
[target_vdu
["cloud-init"]]
1157 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1158 cloud_init_content
=cloud_content_
,
1159 params
=target_vdu
.get("additionalParams"),
1160 context
=target_vdu
["cloud-init"],
1163 if target_vdu
.get("boot-data-drive"):
1164 cloud_config
["boot-data-drive"] = target_vdu
.get("boot-data-drive")
1169 def _check_vld_information_of_interfaces(
1170 interface
: dict, ns_preffix
: str, vnf_preffix
: str
1172 """Prepare the net_text by the virtual link information for vnf and ns level.
1174 interface (dict): Interface details
1175 ns_preffix (str): Prefix of NS
1176 vnf_preffix (str): Prefix of VNF
1179 net_text (str): information of net
1183 if interface
.get("ns-vld-id"):
1184 net_text
= ns_preffix
+ ":vld." + interface
["ns-vld-id"]
1185 elif interface
.get("vnf-vld-id"):
1186 net_text
= vnf_preffix
+ ":vld." + interface
["vnf-vld-id"]
1191 def _prepare_interface_port_security(interface
: dict) -> None:
1195 interface (dict): Interface details
1198 if "port-security-enabled" in interface
:
1199 interface
["port_security"] = interface
.pop("port-security-enabled")
1201 if "port-security-disable-strategy" in interface
:
1202 interface
["port_security_disable_strategy"] = interface
.pop(
1203 "port-security-disable-strategy"
1207 def _create_net_item_of_interface(interface
: dict, net_text
: str) -> dict:
1208 """Prepare net item including name, port security, floating ip etc.
1211 interface (dict): Interface details
1212 net_text (str): information of net
1215 net_item (dict): Dict including net details
1221 for x
, v
in interface
.items()
1227 "port_security_disable_strategy",
1231 net_item
["net_id"] = "TASK-" + net_text
1232 net_item
["type"] = "virtual"
1237 def _prepare_type_of_interface(
1238 interface
: dict, tasks_by_target_record_id
: dict, net_text
: str, net_item
: dict
1240 """Fill the net item type by interface type such as SR-IOV, OM-MGMT, bridge etc.
1243 interface (dict): Interface details
1244 tasks_by_target_record_id (dict): Task details
1245 net_text (str): information of net
1246 net_item (dict): Dict including net details
1249 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1250 # TODO floating_ip: True/False (or it can be None)
1252 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1253 # Mark the net create task as type data
1255 tasks_by_target_record_id
,
1261 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"][
1265 net_item
["use"] = "data"
1266 net_item
["model"] = interface
["type"]
1267 net_item
["type"] = interface
["type"]
1270 interface
.get("type") == "OM-MGMT"
1271 or interface
.get("mgmt-interface")
1272 or interface
.get("mgmt-vnf")
1274 net_item
["use"] = "mgmt"
1277 # If interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1278 net_item
["use"] = "bridge"
1279 net_item
["model"] = interface
.get("type")
1282 def _prepare_vdu_interfaces(
1288 tasks_by_target_record_id
: dict,
1291 """Prepare the net_item and add net_list, add mgmt interface to extra_dict.
1294 target_vdu (dict): VDU to be created
1295 extra_dict (dict): Dictionary to be filled
1296 ns_preffix (str): NS prefix as string
1297 vnf_preffix (str): VNF prefix as string
1298 logger (object): Logger Object
1299 tasks_by_target_record_id (dict): Task details
1300 net_list (list): Net list of VDU
1302 for iface_index
, interface
in enumerate(target_vdu
["interfaces"]):
1303 net_text
= Ns
._check
_vld
_information
_of
_interfaces
(
1304 interface
, ns_preffix
, vnf_preffix
1307 # Interface not connected to any vld
1309 "Interface {} from vdu {} not connected to any vld".format(
1310 iface_index
, target_vdu
["vdu-name"]
1315 extra_dict
["depends_on"].append(net_text
)
1317 Ns
._prepare
_interface
_port
_security
(interface
)
1319 net_item
= Ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
1321 Ns
._prepare
_type
_of
_interface
(
1322 interface
, tasks_by_target_record_id
, net_text
, net_item
1325 if interface
.get("ip-address"):
1326 net_item
["ip_address"] = interface
["ip-address"]
1328 if interface
.get("mac-address"):
1329 net_item
["mac_address"] = interface
["mac-address"]
1331 net_list
.append(net_item
)
1333 if interface
.get("mgmt-vnf"):
1334 extra_dict
["mgmt_vnf_interface"] = iface_index
1335 elif interface
.get("mgmt-interface"):
1336 extra_dict
["mgmt_vdu_interface"] = iface_index
1339 def _prepare_vdu_ssh_keys(
1340 target_vdu
: dict, ro_nsr_public_key
: dict, cloud_config
: dict
1342 """Add ssh keys to cloud config.
1345 target_vdu (dict): Details of VDU to be created
1346 ro_nsr_public_key (dict): RO NSR public Key
1347 cloud_config (dict): Cloud config details
1352 if target_vdu
.get("ssh-keys"):
1353 ssh_keys
+= target_vdu
.get("ssh-keys")
1355 if target_vdu
.get("ssh-access-required"):
1356 ssh_keys
.append(ro_nsr_public_key
)
1359 cloud_config
["key-pairs"] = ssh_keys
1362 def _select_persistent_root_disk(vsd
: dict, vdu
: dict) -> dict:
1363 """Selects the persistent root disk if exists.
1365 vsd (dict): Virtual storage descriptors in VNFD
1366 vdu (dict): VNF descriptor
1369 root_disk (dict): Selected persistent root disk
1371 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
1375 ) == "persistent-storage:persistent-storage" and root_disk
.get(
1381 def _add_persistent_root_disk_to_disk_list(
1382 vnfd
: dict, target_vdu
: dict, persistent_root_disk
: dict, disk_list
: list
1384 """Find the persistent root disk and add to disk list.
1387 vnfd (dict): VNF descriptor
1388 target_vdu (dict): Details of VDU to be created
1389 persistent_root_disk (dict): Details of persistent root disk
1390 disk_list (list): Disks of VDU
1393 for vdu
in vnfd
.get("vdu", ()):
1394 if vdu
["name"] == target_vdu
["vdu-name"]:
1395 for vsd
in vnfd
.get("virtual-storage-desc", ()):
1396 root_disk
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
1401 persistent_root_disk
[vsd
["id"]] = {
1402 "image_id": vdu
.get("sw-image-desc"),
1403 "size": root_disk
["size-of-storage"],
1406 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1410 def _add_persistent_ordinary_disks_to_disk_list(
1412 persistent_root_disk
: dict,
1413 persistent_ordinary_disk
: dict,
1416 """Fill the disk list by adding persistent ordinary disks.
1419 target_vdu (dict): Details of VDU to be created
1420 persistent_root_disk (dict): Details of persistent root disk
1421 persistent_ordinary_disk (dict): Details of persistent ordinary disk
1422 disk_list (list): Disks of VDU
1425 if target_vdu
.get("virtual-storages"):
1426 for disk
in target_vdu
["virtual-storages"]:
1428 disk
.get("type-of-storage")
1429 == "persistent-storage:persistent-storage"
1430 and disk
["id"] not in persistent_root_disk
.keys()
1432 persistent_ordinary_disk
[disk
["id"]] = {
1433 "size": disk
["size-of-storage"],
1435 disk_list
.append(persistent_ordinary_disk
[disk
["id"]])
1438 def _prepare_vdu_affinity_group_list(
1439 target_vdu
: dict, extra_dict
: dict, ns_preffix
: str
1440 ) -> List
[Dict
[str, any
]]:
1441 """Process affinity group details to prepare affinity group list.
1444 target_vdu (dict): Details of VDU to be created
1445 extra_dict (dict): Dictionary to be filled
1446 ns_preffix (str): Prefix as string
1450 affinity_group_list (list): Affinity group details
1453 affinity_group_list
= []
1455 if target_vdu
.get("affinity-or-anti-affinity-group-id"):
1456 for affinity_group_id
in target_vdu
["affinity-or-anti-affinity-group-id"]:
1458 affinity_group_text
= (
1459 ns_preffix
+ ":affinity-or-anti-affinity-group." + affinity_group_id
1462 if not isinstance(extra_dict
.get("depends_on"), list):
1463 raise NsException("Invalid extra_dict format.")
1465 extra_dict
["depends_on"].append(affinity_group_text
)
1466 affinity_group
["affinity_group_id"] = "TASK-" + affinity_group_text
1467 affinity_group_list
.append(affinity_group
)
1469 return affinity_group_list
1472 def _process_vdu_params(
1473 target_vdu
: Dict
[str, Any
],
1474 indata
: Dict
[str, Any
],
1475 vim_info
: Dict
[str, Any
],
1476 target_record_id
: str,
1477 **kwargs
: Dict
[str, Any
],
1478 ) -> Dict
[str, Any
]:
1479 """Function to process VDU parameters.
1482 target_vdu (Dict[str, Any]): [description]
1483 indata (Dict[str, Any]): [description]
1484 vim_info (Dict[str, Any]): [description]
1485 target_record_id (str): [description]
1488 Dict[str, Any]: [description]
1490 vnfr_id
= kwargs
.get("vnfr_id")
1491 nsr_id
= kwargs
.get("nsr_id")
1492 vnfr
= kwargs
.get("vnfr")
1493 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1494 tasks_by_target_record_id
= kwargs
.get("tasks_by_target_record_id")
1495 logger
= kwargs
.get("logger")
1496 db
= kwargs
.get("db")
1497 fs
= kwargs
.get("fs")
1498 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1500 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
1501 ns_preffix
= "nsrs:{}".format(nsr_id
)
1502 image_text
= ns_preffix
+ ":image." + target_vdu
["ns-image-id"]
1503 flavor_text
= ns_preffix
+ ":flavor." + target_vdu
["ns-flavor-id"]
1504 extra_dict
= {"depends_on": [image_text
, flavor_text
]}
1507 persistent_root_disk
= {}
1508 persistent_ordinary_disk
= {}
1509 vdu_instantiation_volumes_list
= []
1511 vnfd_id
= vnfr
["vnfd-id"]
1512 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
1514 # If the position info is provided for all the interfaces, it will be sorted
1515 # according to position number ascendingly.
1517 True if i
.get("position") is not None else False
1518 for i
in target_vdu
["interfaces"]
1520 Ns
._sort
_vdu
_interfaces
(target_vdu
)
1522 # If the position info is provided for some interfaces but not all of them, the interfaces
1523 # which has specific position numbers will be placed and others' positions will not be taken care.
1525 Ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
1527 # If the position info is not provided for the interfaces, interfaces will be attached
1528 # according to the order in the VNFD.
1529 Ns
._prepare
_vdu
_interfaces
(
1535 tasks_by_target_record_id
,
1540 cloud_config
= Ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
1542 # Prepare VDU ssh keys
1543 Ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
1545 if target_vdu
.get("additionalParams"):
1546 vdu_instantiation_volumes_list
= (
1547 target_vdu
.get("additionalParams").get("OSM").get("vdu_volumes")
1550 if vdu_instantiation_volumes_list
:
1551 # Find the root volumes and add to the disk_list
1552 persistent_root_disk
= Ns
.find_persistent_root_volumes(
1553 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
1556 # Find the ordinary volumes which are not added to the persistent_root_disk
1557 # and put them to the disk list
1558 Ns
.find_persistent_volumes(
1559 persistent_root_disk
,
1561 vdu_instantiation_volumes_list
,
1566 # Vdu_instantiation_volumes_list is empty
1567 # First get add the persistent root disks to disk_list
1568 Ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
1569 vnfd
, target_vdu
, persistent_root_disk
, disk_list
1571 # Add the persistent non-root disks to disk_list
1572 Ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
1573 target_vdu
, persistent_root_disk
, persistent_ordinary_disk
, disk_list
1576 affinity_group_list
= Ns
._prepare
_vdu
_affinity
_group
_list
(
1577 target_vdu
, extra_dict
, ns_preffix
1580 extra_dict
["params"] = {
1581 "name": "{}-{}-{}-{}".format(
1582 indata
["name"][:16],
1583 vnfr
["member-vnf-index-ref"][:16],
1584 target_vdu
["vdu-name"][:32],
1585 target_vdu
.get("count-index") or 0,
1587 "description": target_vdu
["vdu-name"],
1589 "image_id": "TASK-" + image_text
,
1590 "flavor_id": "TASK-" + flavor_text
,
1591 "affinity_group_list": affinity_group_list
,
1592 "net_list": net_list
,
1593 "cloud_config": cloud_config
or None,
1594 "disk_list": disk_list
,
1595 "availability_zone_index": None, # TODO
1596 "availability_zone_list": None, # TODO
1602 def _process_affinity_group_params(
1603 target_affinity_group
: Dict
[str, Any
],
1604 indata
: Dict
[str, Any
],
1605 vim_info
: Dict
[str, Any
],
1606 target_record_id
: str,
1607 **kwargs
: Dict
[str, Any
],
1608 ) -> Dict
[str, Any
]:
1609 """Get affinity or anti-affinity group parameters.
1612 target_affinity_group (Dict[str, Any]): [description]
1613 indata (Dict[str, Any]): [description]
1614 vim_info (Dict[str, Any]): [description]
1615 target_record_id (str): [description]
1618 Dict[str, Any]: [description]
1622 affinity_group_data
= {
1623 "name": target_affinity_group
["name"],
1624 "type": target_affinity_group
["type"],
1625 "scope": target_affinity_group
["scope"],
1628 if target_affinity_group
.get("vim-affinity-group-id"):
1629 affinity_group_data
["vim-affinity-group-id"] = target_affinity_group
[
1630 "vim-affinity-group-id"
1633 extra_dict
["params"] = {
1634 "affinity_group_data": affinity_group_data
,
1640 def _process_recreate_vdu_params(
1641 existing_vdu
: Dict
[str, Any
],
1642 db_nsr
: Dict
[str, Any
],
1643 vim_info
: Dict
[str, Any
],
1644 target_record_id
: str,
1646 **kwargs
: Dict
[str, Any
],
1647 ) -> Dict
[str, Any
]:
1648 """Function to process VDU parameters to recreate.
1651 existing_vdu (Dict[str, Any]): [description]
1652 db_nsr (Dict[str, Any]): [description]
1653 vim_info (Dict[str, Any]): [description]
1654 target_record_id (str): [description]
1655 target_id (str): [description]
1658 Dict[str, Any]: [description]
1660 vnfr
= kwargs
.get("vnfr")
1661 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1662 # logger = kwargs.get("logger")
1663 db
= kwargs
.get("db")
1664 fs
= kwargs
.get("fs")
1665 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1671 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
1672 if vim_details_text
:
1673 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
1675 for iface_index
, interface
in enumerate(existing_vdu
["interfaces"]):
1676 if "port-security-enabled" in interface
:
1677 interface
["port_security"] = interface
.pop("port-security-enabled")
1679 if "port-security-disable-strategy" in interface
:
1680 interface
["port_security_disable_strategy"] = interface
.pop(
1681 "port-security-disable-strategy"
1686 for x
, v
in interface
.items()
1692 "port_security_disable_strategy",
1696 existing_ifaces
= existing_vdu
["vim_info"][target_id
].get(
1697 "interfaces_backup", []
1702 for i
in existing_ifaces
1703 if i
["ip_address"] == interface
["ip-address"]
1708 net_item
["net_id"] = net_id
1709 net_item
["type"] = "virtual"
1711 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1712 # TODO floating_ip: True/False (or it can be None)
1713 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1714 net_item
["use"] = "data"
1715 net_item
["model"] = interface
["type"]
1716 net_item
["type"] = interface
["type"]
1718 interface
.get("type") == "OM-MGMT"
1719 or interface
.get("mgmt-interface")
1720 or interface
.get("mgmt-vnf")
1722 net_item
["use"] = "mgmt"
1724 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1725 net_item
["use"] = "bridge"
1726 net_item
["model"] = interface
.get("type")
1728 if interface
.get("ip-address"):
1729 net_item
["ip_address"] = interface
["ip-address"]
1731 if interface
.get("mac-address"):
1732 net_item
["mac_address"] = interface
["mac-address"]
1734 net_list
.append(net_item
)
1736 if interface
.get("mgmt-vnf"):
1737 extra_dict
["mgmt_vnf_interface"] = iface_index
1738 elif interface
.get("mgmt-interface"):
1739 extra_dict
["mgmt_vdu_interface"] = iface_index
1744 if existing_vdu
.get("cloud-init"):
1745 if existing_vdu
["cloud-init"] not in vdu2cloud_init
:
1746 vdu2cloud_init
[existing_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1749 location
=existing_vdu
["cloud-init"],
1752 cloud_content_
= vdu2cloud_init
[existing_vdu
["cloud-init"]]
1753 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1754 cloud_init_content
=cloud_content_
,
1755 params
=existing_vdu
.get("additionalParams"),
1756 context
=existing_vdu
["cloud-init"],
1759 if existing_vdu
.get("boot-data-drive"):
1760 cloud_config
["boot-data-drive"] = existing_vdu
.get("boot-data-drive")
1764 if existing_vdu
.get("ssh-keys"):
1765 ssh_keys
+= existing_vdu
.get("ssh-keys")
1767 if existing_vdu
.get("ssh-access-required"):
1768 ssh_keys
.append(ro_nsr_public_key
)
1771 cloud_config
["key-pairs"] = ssh_keys
1774 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
1775 disk_list
.append({"vim_id": vol_id
["id"]})
1777 affinity_group_list
= []
1779 if existing_vdu
.get("affinity-or-anti-affinity-group-id"):
1781 for affinity_group_id
in existing_vdu
["affinity-or-anti-affinity-group-id"]:
1782 for group
in db_nsr
.get("affinity-or-anti-affinity-group"):
1784 group
["id"] == affinity_group_id
1785 and group
["vim_info"][target_id
].get("vim_id", None) is not None
1787 affinity_group
["affinity_group_id"] = group
["vim_info"][
1789 ].get("vim_id", None)
1790 affinity_group_list
.append(affinity_group
)
1792 extra_dict
["params"] = {
1793 "name": "{}-{}-{}-{}".format(
1794 db_nsr
["name"][:16],
1795 vnfr
["member-vnf-index-ref"][:16],
1796 existing_vdu
["vdu-name"][:32],
1797 existing_vdu
.get("count-index") or 0,
1799 "description": existing_vdu
["vdu-name"],
1801 "image_id": vim_details
["image"]["id"],
1802 "flavor_id": vim_details
["flavor"]["id"],
1803 "affinity_group_list": affinity_group_list
,
1804 "net_list": net_list
,
1805 "cloud_config": cloud_config
or None,
1806 "disk_list": disk_list
,
1807 "availability_zone_index": None, # TODO
1808 "availability_zone_list": None, # TODO
1813 def calculate_diff_items(
1820 tasks_by_target_record_id
,
1827 """Function that returns the incremental changes (creation, deletion)
1828 related to a specific item `item` to be done. This function should be
1829 called for NS instantiation, NS termination, NS update to add a new VNF
1830 or a new VLD, remove a VNF or VLD, etc.
1831 Item can be `net`, `flavor`, `image` or `vdu`.
1832 It takes a list of target items from indata (which came from the REST API)
1833 and compares with the existing items from db_ro_nsr, identifying the
1834 incremental changes to be done. During the comparison, it calls the method
1835 `process_params` (which was passed as parameter, and is particular for each
1839 indata (Dict[str, Any]): deployment info
1840 db_nsr: NSR record from DB
1841 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1842 db_nsr_update (Dict[str, Any]): NSR info to update in DB
1843 item (str): element to process (net, vdu...)
1844 tasks_by_target_record_id (Dict[str, Any]):
1845 [<target_record_id>, <task>]
1846 action_id (str): action id
1847 nsr_id (str): NSR id
1848 task_index (number): task index to add to task name
1849 vnfr_id (str): VNFR id
1850 vnfr (Dict[str, Any]): VNFR info
1853 List: list with the incremental changes (deletes, creates) for each item
1854 number: current task index
1862 process_params
= None
1863 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
1864 ro_nsr_public_key
= db_ro_nsr
["public_key"]
1866 # According to the type of item, the path, the target_list,
1867 # the existing_list and the method to process params are set
1868 db_path
= self
.db_path_map
[item
]
1869 process_params
= self
.process_params_function_map
[item
]
1870 if item
in ("net", "vdu"):
1871 # This case is specific for the NS VLD (not applied to VDU)
1873 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1874 target_list
= indata
.get("ns", []).get(db_path
, [])
1875 existing_list
= db_nsr
.get(db_path
, [])
1876 # This case is common for VNF VLDs and VNF VDUs
1878 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
1880 (vnf
for vnf
in indata
.get("vnf", ()) if vnf
["_id"] == vnfr_id
),
1883 target_list
= target_vnf
.get(db_path
, []) if target_vnf
else []
1884 existing_list
= vnfr
.get(db_path
, [])
1885 elif item
in ("image", "flavor", "affinity-or-anti-affinity-group"):
1886 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1887 target_list
= indata
.get(item
, [])
1888 existing_list
= db_nsr
.get(item
, [])
1890 raise NsException("Item not supported: {}", item
)
1892 # ensure all the target_list elements has an "id". If not assign the index as id
1893 if target_list
is None:
1895 for target_index
, tl
in enumerate(target_list
):
1896 if tl
and not tl
.get("id"):
1897 tl
["id"] = str(target_index
)
1899 # step 1 items (networks,vdus,...) to be deleted/updated
1900 for item_index
, existing_item
in enumerate(existing_list
):
1902 (t
for t
in target_list
if t
["id"] == existing_item
["id"]),
1906 for target_vim
, existing_viminfo
in existing_item
.get(
1909 if existing_viminfo
is None:
1913 target_viminfo
= target_item
.get("vim_info", {}).get(target_vim
)
1915 target_viminfo
= None
1917 if target_viminfo
is None:
1919 self
._assign
_vim
(target_vim
)
1920 target_record_id
= "{}.{}".format(db_record
, existing_item
["id"])
1923 if target_vim
.startswith("sdn") or target_vim
.startswith("wim"):
1924 # item must be sdn-net instead of net if target_vim is a sdn
1926 target_record_id
+= ".sdn"
1929 "action_id": action_id
,
1931 "task_index": task_index
,
1936 "deployment_info": deployment_info
,
1937 "target_id": target_vim
,
1940 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
1941 "target_record_id": target_record_id
,
1946 # step 2 items (networks,vdus,...) to be created
1947 for target_item
in target_list
:
1950 for item_index
, existing_item
in enumerate(existing_list
):
1951 if existing_item
["id"] == target_item
["id"]:
1955 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
1956 existing_list
.append(target_item
)
1957 existing_item
= None
1959 for target_vim
, target_viminfo
in target_item
.get("vim_info", {}).items():
1960 existing_viminfo
= None
1963 existing_viminfo
= existing_item
.get("vim_info", {}).get(target_vim
)
1965 if existing_viminfo
is not None:
1968 target_record_id
= "{}.{}".format(db_record
, target_item
["id"])
1971 if target_vim
.startswith("sdn") or target_vim
.startswith("wim"):
1972 # item must be sdn-net instead of net if target_vim is a sdn
1974 target_record_id
+= ".sdn"
1977 self
.logger
.warning(
1978 "ns.calculate_diff_items target_item={}".format(target_item
)
1980 if process_params
== Ns
._process
_flavor
_params
:
1986 self
.logger
.warning(
1987 "calculate_diff_items for flavor kwargs={}".format(kwargs
)
1990 if process_params
== Ns
._process
_vdu
_params
:
1991 self
.logger
.warning(
1992 "calculate_diff_items self.fs={}".format(self
.fs
)
1999 "vdu2cloud_init": vdu2cloud_init
,
2000 "tasks_by_target_record_id": tasks_by_target_record_id
,
2001 "logger": self
.logger
,
2004 "ro_nsr_public_key": ro_nsr_public_key
,
2007 self
.logger
.warning("calculate_diff_items kwargs={}".format(kwargs
))
2009 extra_dict
= process_params(
2016 self
._assign
_vim
(target_vim
)
2019 "action_id": action_id
,
2021 "task_index": task_index
,
2025 "deployment_info": deployment_info
,
2026 "target_id": target_vim
,
2029 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
2030 "target_record_id": target_record_id
,
2031 "extra_dict": extra_dict
,
2032 "common_id": target_item
.get("common_id", None),
2034 diff_items
.append(new_item
)
2035 tasks_by_target_record_id
[target_record_id
] = new_item
2038 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
2040 return diff_items
, task_index
2042 def calculate_all_differences_to_deploy(
2052 tasks_by_target_record_id
,
2054 """This method calculates the ordered list of items (`changes_list`)
2055 to be created and deleted.
2058 indata (Dict[str, Any]): deployment info
2059 nsr_id (str): NSR id
2060 db_nsr: NSR record from DB
2061 db_vnfrs: VNFRS record from DB
2062 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2063 db_nsr_update (Dict[str, Any]): NSR info to update in DB
2064 db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
2065 action_id (str): action id
2066 tasks_by_target_record_id (Dict[str, Any]):
2067 [<target_record_id>, <task>]
2070 List: ordered list of items to be created and deleted.
2074 # set list with diffs:
2077 # NS vld, image and flavor
2078 for item
in ["net", "image", "flavor", "affinity-or-anti-affinity-group"]:
2079 self
.logger
.debug("process NS={} {}".format(nsr_id
, item
))
2080 diff_items
, task_index
= self
.calculate_diff_items(
2083 db_ro_nsr
=db_ro_nsr
,
2084 db_nsr_update
=db_nsr_update
,
2086 tasks_by_target_record_id
=tasks_by_target_record_id
,
2087 action_id
=action_id
,
2089 task_index
=task_index
,
2092 changes_list
+= diff_items
2095 for vnfr_id
, vnfr
in db_vnfrs
.items():
2096 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
2097 for item
in ["net", "vdu"]:
2098 self
.logger
.debug("process VNF={} {}".format(vnfr_id
, item
))
2099 diff_items
, task_index
= self
.calculate_diff_items(
2102 db_ro_nsr
=db_ro_nsr
,
2103 db_nsr_update
=db_vnfrs_update
[vnfr
["_id"]],
2105 tasks_by_target_record_id
=tasks_by_target_record_id
,
2106 action_id
=action_id
,
2108 task_index
=task_index
,
2112 changes_list
+= diff_items
2116 def define_all_tasks(
2120 tasks_by_target_record_id
,
2122 """Function to create all the task structures obtanied from
2123 the method calculate_all_differences_to_deploy
2126 changes_list (List): ordered list of items to be created or deleted
2127 db_new_tasks (List): tasks list to be created
2128 action_id (str): action id
2129 tasks_by_target_record_id (Dict[str, Any]):
2130 [<target_record_id>, <task>]
2134 for change
in changes_list
:
2135 task
= Ns
._create
_task
(
2136 deployment_info
=change
["deployment_info"],
2137 target_id
=change
["target_id"],
2138 item
=change
["item"],
2139 action
=change
["action"],
2140 target_record
=change
["target_record"],
2141 target_record_id
=change
["target_record_id"],
2142 extra_dict
=change
.get("extra_dict", None),
2145 self
.logger
.warning("ns.define_all_tasks task={}".format(task
))
2146 tasks_by_target_record_id
[change
["target_record_id"]] = task
2147 db_new_tasks
.append(task
)
2149 if change
.get("common_id"):
2150 task
["common_id"] = change
["common_id"]
2152 def upload_all_tasks(
2157 """Function to save all tasks in the common DB
2160 db_new_tasks (List): tasks list to be created
2161 now (time): current time
2165 nb_ro_tasks
= 0 # for logging
2167 for db_task
in db_new_tasks
:
2168 target_id
= db_task
.pop("target_id")
2169 common_id
= db_task
.get("common_id")
2171 # Do not chek tasks with vim_status DELETED
2172 # because in manual heealing there are two tasks for the same vdur:
2173 # one with vim_status deleted and the other one with the actual VM status.
2179 "target_id": target_id
,
2180 "tasks.common_id": common_id
,
2181 "vim_info.vim_status.ne": "DELETED",
2183 update_dict
={"to_check_at": now
, "modified_at": now
},
2184 push
={"tasks": db_task
},
2185 fail_on_empty
=False,
2189 if not self
.db
.set_one(
2192 "target_id": target_id
,
2193 "tasks.target_record": db_task
["target_record"],
2194 "vim_info.vim_status.ne": "DELETED",
2196 update_dict
={"to_check_at": now
, "modified_at": now
},
2197 push
={"tasks": db_task
},
2198 fail_on_empty
=False,
2201 self
.logger
.debug("Updating database, Creating ro_tasks")
2202 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
2204 self
.db
.create("ro_tasks", db_ro_task
)
2207 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2208 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
2212 def upload_recreate_tasks(
2217 """Function to save recreate tasks in the common DB
2220 db_new_tasks (List): tasks list to be created
2221 now (time): current time
2225 nb_ro_tasks
= 0 # for logging
2227 for db_task
in db_new_tasks
:
2228 target_id
= db_task
.pop("target_id")
2229 self
.logger
.warning("target_id={} db_task={}".format(target_id
, db_task
))
2231 action
= db_task
.get("action", None)
2234 self
.logger
.debug("Updating database, Creating ro_tasks")
2235 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
2237 # If DELETE task: the associated created items should be removed
2238 # (except persistent volumes):
2239 if action
== "DELETE":
2240 db_ro_task
["vim_info"]["created"] = True
2241 db_ro_task
["vim_info"]["created_items"] = db_task
.get(
2244 db_ro_task
["vim_info"]["volumes_to_hold"] = db_task
.get(
2245 "volumes_to_hold", []
2247 db_ro_task
["vim_info"]["vim_id"] = db_task
.get("vim_id", None)
2250 self
.logger
.warning("upload_all_tasks db_ro_task={}".format(db_ro_task
))
2251 self
.db
.create("ro_tasks", db_ro_task
)
2254 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2255 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
2259 def _prepare_created_items_for_healing(
2265 # Get created_items from ro_task
2266 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2267 for ro_task
in ro_tasks
:
2268 for task
in ro_task
["tasks"]:
2270 task
["target_record"] == target_record
2271 and task
["action"] == "CREATE"
2272 and ro_task
["vim_info"]["created_items"]
2274 created_items
= ro_task
["vim_info"]["created_items"]
2277 return created_items
2279 def _prepare_persistent_volumes_for_healing(
2284 # The associated volumes of the VM shouldn't be removed
2287 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
2288 if vim_details_text
:
2289 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
2291 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
2292 volumes_list
.append(vol_id
["id"])
2296 def prepare_changes_to_recreate(
2304 tasks_by_target_record_id
,
2306 """This method will obtain an ordered list of items (`changes_list`)
2307 to be created and deleted to meet the recreate request.
2311 "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id
, indata
)
2315 # set list with diffs:
2317 db_path
= self
.db_path_map
["vdu"]
2318 target_list
= indata
.get("healVnfData", {})
2319 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
2320 ro_nsr_public_key
= db_ro_nsr
["public_key"]
2322 # Check each VNF of the target
2323 for target_vnf
in target_list
:
2324 # Find this VNF in the list from DB, raise exception if vnfInstanceId is not found
2325 vnfr_id
= target_vnf
["vnfInstanceId"]
2326 existing_vnf
= db_vnfrs
.get(vnfr_id
)
2327 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
2328 # vim_account_id = existing_vnf.get("vim-account-id", "")
2330 target_vdus
= target_vnf
.get("additionalParams", {}).get("vdu", [])
2331 # Check each VDU of this VNF
2333 # Create target_vdu_list from DB, if VDUs are not specified
2335 for existing_vdu
in existing_vnf
.get("vdur"):
2336 vdu_name
= existing_vdu
.get("vdu-name", None)
2337 vdu_index
= existing_vdu
.get("count-index", 0)
2338 vdu_to_be_healed
= {"vdu-id": vdu_name
, "count-index": vdu_index
}
2339 target_vdus
.append(vdu_to_be_healed
)
2340 for target_vdu
in target_vdus
:
2341 vdu_name
= target_vdu
.get("vdu-id", None)
2342 # For multi instance VDU count-index is mandatory
2343 # For single session VDU count-indes is 0
2344 count_index
= target_vdu
.get("count-index", 0)
2346 existing_instance
= None
2347 for instance
in existing_vnf
.get("vdur", None):
2349 instance
["vdu-name"] == vdu_name
2350 and instance
["count-index"] == count_index
2352 existing_instance
= instance
2357 target_record_id
= "{}.{}".format(db_record
, existing_instance
["id"])
2359 # The target VIM is the one already existing in DB to recreate
2360 for target_vim
, target_viminfo
in existing_instance
.get(
2363 # step 1 vdu to be deleted
2364 self
._assign
_vim
(target_vim
)
2366 "action_id": action_id
,
2368 "task_index": task_index
,
2371 target_record
= f
"{db_record}.{item_index}.vim_info.{target_vim}"
2372 created_items
= self
._prepare
_created
_items
_for
_healing
(
2373 nsr_id
, target_record
2376 volumes_to_hold
= self
._prepare
_persistent
_volumes
_for
_healing
(
2377 target_vim
, existing_instance
2380 # Specific extra params for recreate tasks:
2382 "created_items": created_items
,
2383 "vim_id": existing_instance
["vim-id"],
2384 "volumes_to_hold": volumes_to_hold
,
2387 changes_list
.append(
2389 "deployment_info": deployment_info
,
2390 "target_id": target_vim
,
2393 "target_record": target_record
,
2394 "target_record_id": target_record_id
,
2395 "extra_dict": extra_dict
,
2398 delete_task_id
= f
"{action_id}:{task_index}"
2401 # step 2 vdu to be created
2407 "vnfr": existing_vnf
,
2408 "vdu2cloud_init": vdu2cloud_init
,
2409 "tasks_by_target_record_id": tasks_by_target_record_id
,
2410 "logger": self
.logger
,
2413 "ro_nsr_public_key": ro_nsr_public_key
,
2417 extra_dict
= self
._process
_recreate
_vdu
_params
(
2426 # The CREATE task depens on the DELETE task
2427 extra_dict
["depends_on"] = [delete_task_id
]
2429 # Add volumes created from created_items if any
2430 # Ports should be deleted with delete task and automatically created with create task
2432 for k
, v
in created_items
.items():
2434 k_item
, _
, k_id
= k
.partition(":")
2435 if k_item
== "volume":
2437 except Exception as e
:
2439 "Error evaluating created item {}: {}".format(k
, e
)
2441 extra_dict
["previous_created_volumes"] = volumes
2444 "action_id": action_id
,
2446 "task_index": task_index
,
2448 self
._assign
_vim
(target_vim
)
2451 "deployment_info": deployment_info
,
2452 "target_id": target_vim
,
2455 "target_record": target_record
,
2456 "target_record_id": target_record_id
,
2457 "extra_dict": extra_dict
,
2459 changes_list
.append(new_item
)
2460 tasks_by_target_record_id
[target_record_id
] = new_item
2465 def recreate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2466 self
.logger
.debug("ns.recreate nsr_id={} indata={}".format(nsr_id
, indata
))
2467 # TODO: validate_input(indata, recreate_schema)
2468 action_id
= indata
.get("action_id", str(uuid4()))
2469 # get current deployment
2470 db_vnfrs
= {} # vnf's info indexed by _id
2472 logging_text
= "Recreate nsr_id={} action_id={} indata={}".format(
2473 nsr_id
, action_id
, indata
2475 self
.logger
.debug(logging_text
+ "Enter")
2478 step
= "Getting ns and vnfr record from db"
2479 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2481 tasks_by_target_record_id
= {}
2482 # read from db: vnf's of this ns
2483 step
= "Getting vnfrs from db"
2484 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2485 self
.logger
.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list
))
2487 if not db_vnfrs_list
:
2488 raise NsException("Cannot obtain associated VNF for ns")
2490 for vnfr
in db_vnfrs_list
:
2491 db_vnfrs
[vnfr
["_id"]] = vnfr
2494 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2495 self
.logger
.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr
))
2498 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2500 with self
.write_lock
:
2502 step
= "process NS elements"
2503 changes_list
= self
.prepare_changes_to_recreate(
2508 db_ro_nsr
=db_ro_nsr
,
2509 action_id
=action_id
,
2510 tasks_by_target_record_id
=tasks_by_target_record_id
,
2513 self
.define_all_tasks(
2514 changes_list
=changes_list
,
2515 db_new_tasks
=db_new_tasks
,
2516 tasks_by_target_record_id
=tasks_by_target_record_id
,
2519 # Delete all ro_tasks registered for the targets vdurs (target_record)
2520 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2521 # So remove all task related to target record.
2522 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2523 for change
in changes_list
:
2524 for ro_task
in ro_tasks
:
2525 for task
in ro_task
["tasks"]:
2526 if task
["target_record"] == change
["target_record"]:
2530 "_id": ro_task
["_id"],
2531 "modified_at": ro_task
["modified_at"],
2533 fail_on_empty
=False,
2536 step
= "Updating database, Appending tasks to ro_tasks"
2537 self
.upload_recreate_tasks(
2538 db_new_tasks
=db_new_tasks
,
2543 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2547 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2551 except Exception as e
:
2552 if isinstance(e
, (DbException
, NsException
)):
2554 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2557 e
= traceback_format_exc()
2558 self
.logger
.critical(
2559 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2563 raise NsException(e
)
2565 def deploy(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2566 self
.logger
.debug("ns.deploy nsr_id={} indata={}".format(nsr_id
, indata
))
2567 validate_input(indata
, deploy_schema
)
2568 action_id
= indata
.get("action_id", str(uuid4()))
2570 # get current deployment
2571 db_nsr_update
= {} # update operation on nsrs
2572 db_vnfrs_update
= {}
2573 db_vnfrs
= {} # vnf's info indexed by _id
2575 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2576 self
.logger
.debug(logging_text
+ "Enter")
2579 step
= "Getting ns and vnfr record from db"
2580 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2581 self
.logger
.debug("ns.deploy: db_nsr={}".format(db_nsr
))
2583 tasks_by_target_record_id
= {}
2584 # read from db: vnf's of this ns
2585 step
= "Getting vnfrs from db"
2586 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2588 if not db_vnfrs_list
:
2589 raise NsException("Cannot obtain associated VNF for ns")
2591 for vnfr
in db_vnfrs_list
:
2592 db_vnfrs
[vnfr
["_id"]] = vnfr
2593 db_vnfrs_update
[vnfr
["_id"]] = {}
2594 self
.logger
.debug("ns.deploy db_vnfrs={}".format(db_vnfrs
))
2597 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2600 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2602 # check that action_id is not in the list of actions. Suffixed with :index
2603 if action_id
in db_ro_nsr
["actions"]:
2607 new_action_id
= "{}:{}".format(action_id
, index
)
2609 if new_action_id
not in db_ro_nsr
["actions"]:
2610 action_id
= new_action_id
2613 + "Changing action_id in use to {}".format(action_id
)
2619 def _process_action(indata
):
2620 nonlocal db_new_tasks
2627 if indata
["action"]["action"] == "inject_ssh_key":
2628 key
= indata
["action"].get("key")
2629 user
= indata
["action"].get("user")
2630 password
= indata
["action"].get("password")
2632 for vnf
in indata
.get("vnf", ()):
2633 if vnf
["_id"] not in db_vnfrs
:
2634 raise NsException("Invalid vnf={}".format(vnf
["_id"]))
2636 db_vnfr
= db_vnfrs
[vnf
["_id"]]
2638 for target_vdu
in vnf
.get("vdur", ()):
2639 vdu_index
, vdur
= next(
2642 for i_v
in enumerate(db_vnfr
["vdur"])
2643 if i_v
[1]["id"] == target_vdu
["id"]
2650 "Invalid vdu vnf={}.{}".format(
2651 vnf
["_id"], target_vdu
["id"]
2655 target_vim
, vim_info
= next(
2656 k_v
for k_v
in vdur
["vim_info"].items()
2658 self
._assign
_vim
(target_vim
)
2659 target_record
= "vnfrs:{}:vdur.{}.ssh_keys".format(
2660 vnf
["_id"], vdu_index
2664 "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdur
["id"])
2667 "ip_address": vdur
.get("ip-address"),
2670 "password": password
,
2671 "private_key": db_ro_nsr
["private_key"],
2672 "salt": db_ro_nsr
["_id"],
2673 "schema_version": db_ro_nsr
["_admin"][
2680 "action_id": action_id
,
2682 "task_index": task_index
,
2685 task
= Ns
._create
_task
(
2686 deployment_info
=deployment_info
,
2687 target_id
=target_vim
,
2690 target_record
=target_record
,
2691 target_record_id
=None,
2692 extra_dict
=extra_dict
,
2695 task_index
= deployment_info
.get("task_index")
2697 db_new_tasks
.append(task
)
2699 with self
.write_lock
:
2700 if indata
.get("action"):
2701 _process_action(indata
)
2703 # compute network differences
2705 step
= "process NS elements"
2706 changes_list
= self
.calculate_all_differences_to_deploy(
2711 db_ro_nsr
=db_ro_nsr
,
2712 db_nsr_update
=db_nsr_update
,
2713 db_vnfrs_update
=db_vnfrs_update
,
2714 action_id
=action_id
,
2715 tasks_by_target_record_id
=tasks_by_target_record_id
,
2717 self
.define_all_tasks(
2718 changes_list
=changes_list
,
2719 db_new_tasks
=db_new_tasks
,
2720 tasks_by_target_record_id
=tasks_by_target_record_id
,
2723 step
= "Updating database, Appending tasks to ro_tasks"
2724 self
.upload_all_tasks(
2725 db_new_tasks
=db_new_tasks
,
2729 step
= "Updating database, nsrs"
2731 self
.db
.set_one("nsrs", {"_id": nsr_id
}, db_nsr_update
)
2733 for vnfr_id
, db_vnfr_update
in db_vnfrs_update
.items():
2735 step
= "Updating database, vnfrs={}".format(vnfr_id
)
2736 self
.db
.set_one("vnfrs", {"_id": vnfr_id
}, db_vnfr_update
)
2739 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2743 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2747 except Exception as e
:
2748 if isinstance(e
, (DbException
, NsException
)):
2750 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2753 e
= traceback_format_exc()
2754 self
.logger
.critical(
2755 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2759 raise NsException(e
)
2761 def delete(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2762 self
.logger
.debug("ns.delete version={} nsr_id={}".format(version
, nsr_id
))
2763 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2765 with self
.write_lock
:
2767 NsWorker
.delete_db_tasks(self
.db
, nsr_id
, None)
2768 except NsWorkerException
as e
:
2769 raise NsException(e
)
2771 return None, None, True
2773 def status(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2775 "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2776 version
, nsr_id
, action_id
, indata
2782 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.action_id": action_id
})
2783 global_status
= "DONE"
2786 for ro_task
in ro_tasks
:
2787 for task
in ro_task
["tasks"]:
2788 if task
and task
["action_id"] == action_id
:
2789 task_list
.append(task
)
2792 if task
["status"] == "FAILED":
2793 global_status
= "FAILED"
2794 error_text
= "Error at {} {}: {}".format(
2795 task
["action"].lower(),
2797 ro_task
["vim_info"].get("vim_message") or "unknown",
2799 details
.append(error_text
)
2800 elif task
["status"] in ("SCHEDULED", "BUILD"):
2801 if global_status
!= "FAILED":
2802 global_status
= "BUILD"
2807 "status": global_status
,
2808 "details": ". ".join(details
)
2810 else "progress {}/{}".format(done
, total
),
2812 "action_id": action_id
,
2816 return return_data
, None, True
2818 def recreate_status(
2819 self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
2821 return self
.status(session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
)
2823 def cancel(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2825 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2826 session
, indata
, version
, nsr_id
, action_id
2830 return None, None, True
2832 def rebuild_start_stop_task(
2843 self
._assign
_vim
(target_vim
)
2844 target_record
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_index
)
2845 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_id
)
2847 "action_id": action_id
,
2849 "task_index": task_index
,
2852 task
= Ns
._create
_task
(
2853 deployment_info
=deployment_info
,
2854 target_id
=target_vim
,
2857 target_record
=target_record
,
2858 target_record_id
=target_record_id
,
2859 extra_dict
=extra_dict
,
2863 def rebuild_start_stop(
2864 self
, session
, action_dict
, version
, nsr_id
, *args
, **kwargs
2869 action_id
= action_dict
.get("action_id", str(uuid4()))
2871 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2872 self
.logger
.debug(logging_text
+ "Enter")
2874 action
= list(action_dict
.keys())[0]
2875 task_dict
= action_dict
.get(action
)
2876 vim_vm_id
= action_dict
.get(action
).get("vim_vm_id")
2878 if action_dict
.get("stop"):
2882 step
= "lock the operation & do task creation"
2883 with self
.write_lock
:
2884 extra_dict
["params"] = {
2885 "vim_vm_id": vim_vm_id
,
2888 task
= self
.rebuild_start_stop_task(
2889 task_dict
["vdu_id"],
2890 task_dict
["vnf_id"],
2891 task_dict
["vdu_index"],
2895 task_dict
["target_vim"],
2898 db_new_tasks
.append(task
)
2899 step
= "upload Task to db"
2900 self
.upload_all_tasks(
2901 db_new_tasks
=db_new_tasks
,
2905 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2908 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2912 except Exception as e
:
2913 if isinstance(e
, (DbException
, NsException
)):
2915 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2918 e
= traceback_format_exc()
2919 self
.logger
.critical(
2920 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2923 raise NsException(e
)
2925 def get_deploy(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2926 nsrs
= self
.db
.get_list("nsrs", {})
2930 return_data
.append({"_id": ns
["_id"], "name": ns
["name"]})
2932 return return_data
, None, True
2934 def get_actions(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2935 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2938 for ro_task
in ro_tasks
:
2939 for task
in ro_task
["tasks"]:
2940 if task
["action_id"] not in return_data
:
2941 return_data
.append(task
["action_id"])
2943 return return_data
, None, True
2946 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
2948 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
2949 self
._assign
_vim
(target_vim
)
2950 target_record
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu_index
)
2951 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
2953 "action_id": action_id
,
2955 "task_index": task_index
,
2958 task
= Ns
._create
_task
(
2959 deployment_info
=deployment_info
,
2960 target_id
=target_vim
,
2963 target_record
=target_record
,
2964 target_record_id
=target_record_id
,
2965 extra_dict
=extra_dict
,
2970 def migrate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2974 action_id
= indata
.get("action_id", str(uuid4()))
2976 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2977 self
.logger
.debug(logging_text
+ "Enter")
2979 vnf_instance_id
= indata
["vnfInstanceId"]
2980 step
= "Getting vnfrs from db"
2981 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
2982 vdu
= indata
.get("vdu")
2983 migrateToHost
= indata
.get("migrateToHost")
2986 with self
.write_lock
:
2988 vdu_id
= indata
["vdu"]["vduId"]
2989 vdu_count_index
= indata
["vdu"].get("vduCountIndex", 0)
2990 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2992 vdu
["vdu-id-ref"] == vdu_id
2993 and vdu
["count-index"] == vdu_count_index
2995 extra_dict
["params"] = {
2996 "vim_vm_id": vdu
["vim-id"],
2997 "migrate_host": migrateToHost
,
2998 "vdu_vim_info": vdu
["vim_info"],
3000 step
= "Creating migration task for vdu:{}".format(vdu
)
3001 task
= self
.migrate_task(
3010 db_new_tasks
.append(task
)
3014 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3015 extra_dict
["params"] = {
3016 "vim_vm_id": vdu
["vim-id"],
3017 "migrate_host": migrateToHost
,
3018 "vdu_vim_info": vdu
["vim_info"],
3020 step
= "Creating migration task for vdu:{}".format(vdu
)
3021 task
= self
.migrate_task(
3030 db_new_tasks
.append(task
)
3033 self
.upload_all_tasks(
3034 db_new_tasks
=db_new_tasks
,
3039 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3042 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3046 except Exception as e
:
3047 if isinstance(e
, (DbException
, NsException
)):
3049 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3052 e
= traceback_format_exc()
3053 self
.logger
.critical(
3054 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3057 raise NsException(e
)
3059 def verticalscale_task(
3060 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3062 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
3063 self
._assign
_vim
(target_vim
)
3064 target_record
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu_index
)
3065 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
3067 "action_id": action_id
,
3069 "task_index": task_index
,
3072 task
= Ns
._create
_task
(
3073 deployment_info
=deployment_info
,
3074 target_id
=target_vim
,
3075 item
="verticalscale",
3077 target_record
=target_record
,
3078 target_record_id
=target_record_id
,
3079 extra_dict
=extra_dict
,
3083 def verticalscale(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3087 action_id
= indata
.get("action_id", str(uuid4()))
3089 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3090 self
.logger
.debug(logging_text
+ "Enter")
3092 VnfFlavorData
= indata
.get("changeVnfFlavorData")
3093 vnf_instance_id
= VnfFlavorData
["vnfInstanceId"]
3094 step
= "Getting vnfrs from db"
3095 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
3096 vduid
= VnfFlavorData
["additionalParams"]["vduid"]
3097 vduCountIndex
= VnfFlavorData
["additionalParams"]["vduCountIndex"]
3098 virtualMemory
= VnfFlavorData
["additionalParams"]["virtualMemory"]
3099 numVirtualCpu
= VnfFlavorData
["additionalParams"]["numVirtualCpu"]
3100 sizeOfStorage
= VnfFlavorData
["additionalParams"]["sizeOfStorage"]
3102 "name": vduid
+ "-flv",
3103 "ram": virtualMemory
,
3104 "vcpus": numVirtualCpu
,
3105 "disk": sizeOfStorage
,
3108 step
= "Creating Tasks for vertical scaling"
3109 with self
.write_lock
:
3110 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3112 vdu
["vdu-id-ref"] == vduid
3113 and vdu
["count-index"] == vduCountIndex
3115 extra_dict
["params"] = {
3116 "vim_vm_id": vdu
["vim-id"],
3117 "flavor_dict": flavor_dict
,
3119 task
= self
.verticalscale_task(
3128 db_new_tasks
.append(task
)
3131 self
.upload_all_tasks(
3132 db_new_tasks
=db_new_tasks
,
3136 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3139 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3143 except Exception as e
:
3144 if isinstance(e
, (DbException
, NsException
)):
3146 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3149 e
= traceback_format_exc()
3150 self
.logger
.critical(
3151 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3154 raise NsException(e
)