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 copy
import deepcopy
20 from http
import HTTPStatus
21 from itertools
import product
23 from random
import choice
as random_choice
24 from threading
import Lock
26 from traceback
import format_exc
as traceback_format_exc
27 from typing
import Any
, Dict
, List
, Optional
, Tuple
, Type
28 from uuid
import uuid4
30 from cryptography
.hazmat
.backends
import default_backend
as crypto_default_backend
31 from cryptography
.hazmat
.primitives
import serialization
as crypto_serialization
32 from cryptography
.hazmat
.primitives
.asymmetric
import rsa
41 from osm_common
import (
48 version
as common_version
,
50 from osm_common
.dbbase
import DbBase
, DbException
51 from osm_common
.fsbase
import FsBase
, FsException
52 from osm_common
.msgbase
import MsgException
53 from osm_ng_ro
.ns_thread
import deep_get
, NsWorker
, NsWorkerException
54 from osm_ng_ro
.validation
import deploy_schema
, validate_input
57 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
58 min_common_version
= "0.1.16"
61 class NsException(Exception):
62 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
63 self
.http_code
= http_code
64 super(Exception, self
).__init
__(message
)
69 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
70 will provide a random one
73 # Try getting docker id. If fails, get pid
75 with
open("/proc/self/cgroup", "r") as f
:
76 text_id_
= f
.readline()
77 _
, _
, text_id
= text_id_
.rpartition("/")
78 text_id
= text_id
.replace("\n", "")[:12]
82 except Exception as error
:
83 logging
.exception(f
"{error} occured while getting process id")
86 return "".join(random_choice("0123456789abcdef") for _
in range(12))
90 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
93 for point
in v
.split("."):
94 filled
.append(point
.zfill(8))
105 # self.operations = None
107 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
108 # If done now it will not be linked to parent not getting its handler and level
110 self
.write_lock
= None
111 self
.vims_assigned
= {}
115 self
.process_params_function_map
= {
116 "net": Ns
._process
_net
_params
,
117 "image": Ns
._process
_image
_params
,
118 "flavor": Ns
._process
_flavor
_params
,
119 "vdu": Ns
._process
_vdu
_params
,
120 "classification": Ns
._process
_classification
_params
,
121 "sfi": Ns
._process
_sfi
_params
,
122 "sf": Ns
._process
_sf
_params
,
123 "sfp": Ns
._process
_sfp
_params
,
124 "affinity-or-anti-affinity-group": Ns
._process
_affinity
_group
_params
,
125 "shared-volumes": Ns
._process
_shared
_volumes
_params
,
132 "classification": "classification",
136 "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
137 "shared-volumes": "shared-volumes",
140 def init_db(self
, target_version
):
143 def start(self
, config
):
145 Connect to database, filesystem storage, and messaging
146 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
147 :param config: Configuration of db, storage, etc
151 self
.config
["process_id"] = get_process_id() # used for HA identity
152 self
.logger
= logging
.getLogger("ro.ns")
154 # check right version of common
155 if versiontuple(common_version
) < versiontuple(min_common_version
):
157 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
158 common_version
, min_common_version
164 if config
["database"]["driver"] == "mongo":
165 self
.db
= dbmongo
.DbMongo()
166 self
.db
.db_connect(config
["database"])
167 elif config
["database"]["driver"] == "memory":
168 self
.db
= dbmemory
.DbMemory()
169 self
.db
.db_connect(config
["database"])
172 "Invalid configuration param '{}' at '[database]':'driver'".format(
173 config
["database"]["driver"]
178 if config
["storage"]["driver"] == "local":
179 self
.fs
= fslocal
.FsLocal()
180 self
.fs
.fs_connect(config
["storage"])
181 elif config
["storage"]["driver"] == "mongo":
182 self
.fs
= fsmongo
.FsMongo()
183 self
.fs
.fs_connect(config
["storage"])
184 elif config
["storage"]["driver"] is None:
188 "Invalid configuration param '{}' at '[storage]':'driver'".format(
189 config
["storage"]["driver"]
194 if config
["message"]["driver"] == "local":
195 self
.msg
= msglocal
.MsgLocal()
196 self
.msg
.connect(config
["message"])
197 elif config
["message"]["driver"] == "kafka":
198 self
.msg
= msgkafka
.MsgKafka()
199 self
.msg
.connect(config
["message"])
202 "Invalid configuration param '{}' at '[message]':'driver'".format(
203 config
["message"]["driver"]
207 # TODO load workers to deal with exising database tasks
209 self
.write_lock
= Lock()
210 except (DbException
, FsException
, MsgException
) as e
:
211 raise NsException(str(e
), http_code
=e
.http_code
)
213 def get_assigned_vims(self
):
214 return list(self
.vims_assigned
.keys())
219 self
.db
.db_disconnect()
222 self
.fs
.fs_disconnect()
225 self
.msg
.disconnect()
227 self
.write_lock
= None
228 except (DbException
, FsException
, MsgException
) as e
:
229 raise NsException(str(e
), http_code
=e
.http_code
)
231 for worker
in self
.workers
:
232 worker
.insert_task(("terminate",))
234 def _create_worker(self
):
236 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
237 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
238 return the index of the assigned worker thread. Worker threads are storead at self.workers
240 # Look for a thread in idle status
244 for i
in range(len(self
.workers
))
245 if self
.workers
[i
] and self
.workers
[i
].idle
250 if worker_id
is not None:
251 # unset idle status to avoid race conditions
252 self
.workers
[worker_id
].idle
= False
254 worker_id
= len(self
.workers
)
256 if worker_id
< self
.config
["global"]["server.ns_threads"]:
257 # create a new worker
259 NsWorker(worker_id
, self
.config
, self
.plugins
, self
.db
)
261 self
.workers
[worker_id
].start()
263 # reached maximum number of threads, assign VIM to an existing one
264 worker_id
= self
.next_worker
265 self
.next_worker
= (self
.next_worker
+ 1) % self
.config
["global"][
271 def assign_vim(self
, target_id
):
272 with self
.write_lock
:
273 return self
._assign
_vim
(target_id
)
275 def _assign_vim(self
, target_id
):
276 if target_id
not in self
.vims_assigned
:
277 worker_id
= self
.vims_assigned
[target_id
] = self
._create
_worker
()
278 self
.workers
[worker_id
].insert_task(("load_vim", target_id
))
280 def reload_vim(self
, target_id
):
281 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
282 # this is because database VIM information is cached for threads working with SDN
283 with self
.write_lock
:
284 for worker
in self
.workers
:
285 if worker
and not worker
.idle
:
286 worker
.insert_task(("reload_vim", target_id
))
288 def unload_vim(self
, target_id
):
289 with self
.write_lock
:
290 return self
._unload
_vim
(target_id
)
292 def _unload_vim(self
, target_id
):
293 if target_id
in self
.vims_assigned
:
294 worker_id
= self
.vims_assigned
[target_id
]
295 self
.workers
[worker_id
].insert_task(("unload_vim", target_id
))
296 del self
.vims_assigned
[target_id
]
298 def check_vim(self
, target_id
):
299 with self
.write_lock
:
300 if target_id
in self
.vims_assigned
:
301 worker_id
= self
.vims_assigned
[target_id
]
303 worker_id
= self
._create
_worker
()
305 worker
= self
.workers
[worker_id
]
306 worker
.insert_task(("check_vim", target_id
))
308 def unload_unused_vims(self
):
309 with self
.write_lock
:
312 for target_id
in self
.vims_assigned
:
313 if not self
.db
.get_one(
316 "target_id": target_id
,
317 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
321 vims_to_unload
.append(target_id
)
323 for target_id
in vims_to_unload
:
324 self
._unload
_vim
(target_id
)
332 """This method reads cloud init from a file.
334 Note: Not used as cloud init content is provided in the http body.
337 db (Type[DbBase]): [description]
338 fs (Type[FsBase]): [description]
339 location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
342 NsException: [description]
343 NsException: [description]
348 vnfd_id
, _
, other
= location
.partition(":")
349 _type
, _
, name
= other
.partition(":")
350 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
353 base_folder
= vnfd
["_admin"]["storage"]
354 cloud_init_file
= "{}/{}/cloud_init/{}".format(
355 base_folder
["folder"], base_folder
["pkg-dir"], name
360 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
365 with fs
.file_open(cloud_init_file
, "r") as ci_file
:
366 cloud_init_content
= ci_file
.read()
368 cloud_init_content
= vnfd
["vdu"][int(name
)]["cloud-init"]
370 raise NsException("Mismatch descriptor for cloud init: {}".format(location
))
372 return cloud_init_content
376 cloud_init_content
: str,
377 params
: Dict
[str, Any
],
380 """Function that processes the cloud init to replace Jinja2 encoded parameters.
383 cloud_init_content (str): [description]
384 params (Dict[str, Any]): [description]
385 context (str): [description]
388 NsException: [description]
389 NsException: [description]
396 undefined
=StrictUndefined
,
397 autoescape
=select_autoescape(default_for_string
=True, default
=True),
399 template
= env
.from_string(cloud_init_content
)
401 return template
.render(params
or {})
402 except UndefinedError
as e
:
404 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
405 "inside the 'additionalParamsForVnf' block".format(e
, context
)
407 except (TemplateError
, TemplateNotFound
) as e
:
409 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
414 def _create_db_ro_nsrs(self
, nsr_id
, now
):
416 key
= rsa
.generate_private_key(
417 backend
=crypto_default_backend(), public_exponent
=65537, key_size
=2048
419 private_key
= key
.private_bytes(
420 crypto_serialization
.Encoding
.PEM
,
421 crypto_serialization
.PrivateFormat
.PKCS8
,
422 crypto_serialization
.NoEncryption(),
424 public_key
= key
.public_key().public_bytes(
425 crypto_serialization
.Encoding
.OpenSSH
,
426 crypto_serialization
.PublicFormat
.OpenSSH
,
428 private_key
= private_key
.decode("utf8")
429 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
430 i
= private_key
.find("\n")
431 private_key
= "-----BEGIN RSA PRIVATE KEY-----" + private_key
[i
:]
432 public_key
= public_key
.decode("utf8")
433 except Exception as e
:
434 raise NsException("Cannot create ssh-keys: {}".format(e
))
436 schema_version
= "1.1"
437 private_key_encrypted
= self
.db
.encrypt(
438 private_key
, schema_version
=schema_version
, salt
=nsr_id
445 "schema_version": schema_version
,
447 "public_key": public_key
,
448 "private_key": private_key_encrypted
,
451 self
.db
.create("ro_nsrs", db_content
)
457 deployment_info
: Dict
[str, Any
],
462 target_record_id
: str,
463 extra_dict
: Dict
[str, Any
] = None,
465 """Function to create task dict from deployment information.
468 deployment_info (Dict[str, Any]): [description]
469 target_id (str): [description]
470 item (str): [description]
471 action (str): [description]
472 target_record (str): [description]
473 target_record_id (str): [description]
474 extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
477 Dict[str, Any]: [description]
480 "target_id": target_id
, # it will be removed before pushing at database
481 "action_id": deployment_info
.get("action_id"),
482 "nsr_id": deployment_info
.get("nsr_id"),
483 "task_id": f
"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
484 "status": "SCHEDULED",
487 "target_record": target_record
,
488 "target_record_id": target_record_id
,
492 task
.update(extra_dict
) # params, find_params, depends_on
494 deployment_info
["task_index"] = deployment_info
.get("task_index", 0) + 1
501 task
: Dict
[str, Any
],
503 """Function to create an RO task from task information.
506 target_id (str): [description]
507 task (Dict[str, Any]): [description]
510 Dict[str, Any]: [description]
514 _id
= task
.get("task_id")
519 "target_id": target_id
,
522 "created_items": None,
539 def _process_image_params(
540 target_image
: Dict
[str, Any
],
541 indata
: Dict
[str, Any
],
542 vim_info
: Dict
[str, Any
],
543 target_record_id
: str,
544 **kwargs
: Dict
[str, Any
],
546 """Function to process VDU image parameters.
549 target_image (Dict[str, Any]): [description]
550 indata (Dict[str, Any]): [description]
551 vim_info (Dict[str, Any]): [description]
552 target_record_id (str): [description]
555 Dict[str, Any]: [description]
559 if target_image
.get("image"):
560 find_params
["filter_dict"] = {"name": target_image
.get("image")}
562 if target_image
.get("vim_image_id"):
563 find_params
["filter_dict"] = {"id": target_image
.get("vim_image_id")}
565 if target_image
.get("image_checksum"):
566 find_params
["filter_dict"] = {
567 "checksum": target_image
.get("image_checksum")
570 return {"find_params": find_params
}
573 def _get_resource_allocation_params(
574 quota_descriptor
: Dict
[str, Any
],
576 """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
580 quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
583 Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
587 if quota_descriptor
.get("limit"):
588 quota
["limit"] = int(quota_descriptor
["limit"])
590 if quota_descriptor
.get("reserve"):
591 quota
["reserve"] = int(quota_descriptor
["reserve"])
593 if quota_descriptor
.get("shares"):
594 quota
["shares"] = int(quota_descriptor
["shares"])
599 def _process_guest_epa_quota_params(
600 guest_epa_quota
: Dict
[str, Any
],
603 """Function to extract the guest epa quota parameters.
606 guest_epa_quota (Dict[str, Any]): [description]
607 epa_vcpu_set (bool): [description]
610 Dict[str, Any]: [description]
614 if guest_epa_quota
.get("cpu-quota") and not epa_vcpu_set
:
615 cpuquota
= Ns
._get
_resource
_allocation
_params
(
616 guest_epa_quota
.get("cpu-quota")
620 result
["cpu-quota"] = cpuquota
622 if guest_epa_quota
.get("mem-quota"):
623 vduquota
= Ns
._get
_resource
_allocation
_params
(
624 guest_epa_quota
.get("mem-quota")
628 result
["mem-quota"] = vduquota
630 if guest_epa_quota
.get("disk-io-quota"):
631 diskioquota
= Ns
._get
_resource
_allocation
_params
(
632 guest_epa_quota
.get("disk-io-quota")
636 result
["disk-io-quota"] = diskioquota
638 if guest_epa_quota
.get("vif-quota"):
639 vifquota
= Ns
._get
_resource
_allocation
_params
(
640 guest_epa_quota
.get("vif-quota")
644 result
["vif-quota"] = vifquota
649 def _process_guest_epa_numa_params(
650 guest_epa_quota
: Dict
[str, Any
],
651 ) -> Tuple
[Dict
[str, Any
], bool]:
655 guest_epa_quota (Dict[str, Any]): [description]
658 Tuple[Dict[str, Any], bool]: [description]
664 if guest_epa_quota
.get("numa-node-policy"):
665 numa_node_policy
= guest_epa_quota
.get("numa-node-policy")
667 if numa_node_policy
.get("node"):
668 for numa_node
in numa_node_policy
["node"]:
670 if numa_node
.get("id"):
671 numa
["id"] = int(numa_node
["id"])
673 if numa_node
.get("vcpu"):
674 for vcpu
in numa_node
.get("vcpu"):
675 vcpu_id
= int(vcpu
.get("id"))
676 vcpu_list
.append(vcpu_id
)
677 numa
["vcpu"] = vcpu_list
679 if numa_node
.get("num-cores"):
680 numa
["cores"] = numa_node
["num-cores"]
683 paired_threads
= numa_node
.get("paired-threads", {})
684 if paired_threads
.get("num-paired-threads"):
685 numa
["paired_threads"] = int(
686 numa_node
["paired-threads"]["num-paired-threads"]
690 if paired_threads
.get("paired-thread-ids"):
691 numa
["paired-threads-id"] = []
693 for pair
in paired_threads
["paired-thread-ids"]:
694 numa
["paired-threads-id"].append(
696 str(pair
["thread-a"]),
697 str(pair
["thread-b"]),
701 if numa_node
.get("num-threads"):
702 numa
["threads"] = int(numa_node
["num-threads"])
705 if numa_node
.get("memory-mb"):
706 numa
["memory"] = max(int(int(numa_node
["memory-mb"]) / 1024), 1)
708 numa_list
.append(numa
)
711 return numa_list
, epa_vcpu_set
714 def _process_guest_epa_cpu_pinning_params(
715 guest_epa_quota
: Dict
[str, Any
],
718 ) -> Tuple
[Dict
[str, Any
], bool]:
722 guest_epa_quota (Dict[str, Any]): [description]
723 vcpu_count (int): [description]
724 epa_vcpu_set (bool): [description]
727 Tuple[Dict[str, Any], bool]: [description]
730 local_epa_vcpu_set
= epa_vcpu_set
733 guest_epa_quota
.get("cpu-pinning-policy") == "DEDICATED"
736 # Pinning policy "REQUIRE" uses threads as host should support SMT architecture
737 # Pinning policy "ISOLATE" uses cores as host should not support SMT architecture
738 # Pinning policy "PREFER" uses threads in case host supports SMT architecture
741 if guest_epa_quota
.get("cpu-thread-pinning-policy") == "ISOLATE"
743 ] = max(vcpu_count
, 1)
744 local_epa_vcpu_set
= True
746 return numa
, local_epa_vcpu_set
749 def _process_epa_params(
750 target_flavor
: Dict
[str, Any
],
755 target_flavor (Dict[str, Any]): [description]
758 Dict[str, Any]: [description]
764 if target_flavor
.get("guest-epa"):
765 guest_epa
= target_flavor
["guest-epa"]
767 numa_list
, epa_vcpu_set
= Ns
._process
_guest
_epa
_numa
_params
(
768 guest_epa_quota
=guest_epa
771 if guest_epa
.get("mempage-size"):
772 extended
["mempage-size"] = guest_epa
.get("mempage-size")
774 if guest_epa
.get("cpu-pinning-policy"):
775 extended
["cpu-pinning-policy"] = guest_epa
.get("cpu-pinning-policy")
777 if guest_epa
.get("cpu-thread-pinning-policy"):
778 extended
["cpu-thread-pinning-policy"] = guest_epa
.get(
779 "cpu-thread-pinning-policy"
782 if guest_epa
.get("numa-node-policy"):
783 if guest_epa
.get("numa-node-policy").get("mem-policy"):
784 extended
["mem-policy"] = guest_epa
.get("numa-node-policy").get(
788 tmp_numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
789 guest_epa_quota
=guest_epa
,
790 vcpu_count
=int(target_flavor
.get("vcpu-count", 1)),
791 epa_vcpu_set
=epa_vcpu_set
,
793 for numa
in numa_list
:
794 numa
.update(tmp_numa
)
797 Ns
._process
_guest
_epa
_quota
_params
(
798 guest_epa_quota
=guest_epa
,
799 epa_vcpu_set
=epa_vcpu_set
,
804 extended
["numas"] = numa_list
809 def _process_flavor_params(
810 target_flavor
: Dict
[str, Any
],
811 indata
: Dict
[str, Any
],
812 vim_info
: Dict
[str, Any
],
813 target_record_id
: str,
814 **kwargs
: Dict
[str, Any
],
819 target_flavor (Dict[str, Any]): [description]
820 indata (Dict[str, Any]): [description]
821 vim_info (Dict[str, Any]): [description]
822 target_record_id (str): [description]
825 Dict[str, Any]: [description]
827 db
= kwargs
.get("db")
830 for vnf
in indata
.get("vnf", []):
831 for vdur
in vnf
.get("vdur", []):
832 if vdur
.get("ns-flavor-id") == target_flavor
.get("id"):
836 target_vdur
.get("additionalParams", {}).get("OSM", {}).get("vim_flavor_id")
838 if vim_flavor_id
: # vim-flavor-id was passed so flavor won't be created
839 return {"find_params": {"vim_flavor_id": vim_flavor_id
}}
842 "disk": int(target_flavor
["storage-gb"]),
843 "ram": int(target_flavor
["memory-mb"]),
844 "vcpus": int(target_flavor
["vcpu-count"]),
847 if db
and isinstance(indata
.get("vnf"), list):
848 vnfd_id
= indata
.get("vnf")[0].get("vnfd-id")
849 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
850 # check if there is persistent root disk
851 for vdu
in vnfd
.get("vdu", ()):
852 if vdu
["name"] == target_vdur
.get("vdu-name"):
853 for vsd
in vnfd
.get("virtual-storage-desc", ()):
854 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
857 root_disk
.get("type-of-storage")
858 == "persistent-storage:persistent-storage"
860 flavor_data
["disk"] = 0
862 for storage
in target_vdur
.get("virtual-storages", []):
864 storage
.get("type-of-storage")
865 == "etsi-nfv-descriptors:ephemeral-storage"
867 flavor_data
["ephemeral"] = int(storage
.get("size-of-storage", 0))
868 elif storage
.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
869 flavor_data
["swap"] = int(storage
.get("size-of-storage", 0))
871 extended
= Ns
._process
_epa
_params
(target_flavor
)
873 flavor_data
["extended"] = extended
875 extra_dict
= {"find_params": {"flavor_data": flavor_data
}}
876 flavor_data_name
= flavor_data
.copy()
877 flavor_data_name
["name"] = target_flavor
["name"]
878 extra_dict
["params"] = {"flavor_data": flavor_data_name
}
882 def _prefix_ip_address(ip_address
):
883 if "/" not in ip_address
:
888 def _process_ip_proto(ip_proto
):
899 def _process_classification_params(
900 target_classification
: Dict
[str, Any
],
901 indata
: Dict
[str, Any
],
902 vim_info
: Dict
[str, Any
],
903 target_record_id
: str,
904 **kwargs
: Dict
[str, Any
],
909 target_classification (Dict[str, Any]): Classification dictionary parameters that needs to be processed to create resource on VIM
910 indata (Dict[str, Any]): Deployment info
911 vim_info (Dict[str, Any]):To add items created by OSM on the VIM.
912 target_record_id (str): Task record ID.
913 **kwargs (Dict[str, Any]): Used to send additional information to the task.
916 Dict[str, Any]: Return parameters required to create classification and Items on which classification is dependent.
918 vnfr_id
= target_classification
["vnfr_id"]
919 vdur_id
= target_classification
["vdur_id"]
920 port_index
= target_classification
["ingress_port_index"]
923 classification_data
= {
924 "name": target_classification
["id"],
925 "source_port_range_min": target_classification
["source-port"],
926 "source_port_range_max": target_classification
["source-port"],
927 "destination_port_range_min": target_classification
["destination-port"],
928 "destination_port_range_max": target_classification
["destination-port"],
931 classification_data
["source_ip_prefix"] = Ns
._prefix
_ip
_address
(
932 target_classification
["source-ip-address"]
935 classification_data
["destination_ip_prefix"] = Ns
._prefix
_ip
_address
(
936 target_classification
["destination-ip-address"]
939 classification_data
["protocol"] = Ns
._process
_ip
_proto
(
940 int(target_classification
["ip-proto"])
943 db
= kwargs
.get("db")
944 vdu_text
= Ns
._get
_vnfr
_vdur
_text
(db
, vnfr_id
, vdur_id
)
946 extra_dict
= {"depends_on": [vdu_text
]}
948 extra_dict
= {"depends_on": [vdu_text
]}
949 classification_data
["logical_source_port"] = "TASK-" + vdu_text
950 classification_data
["logical_source_port_index"] = port_index
952 extra_dict
["params"] = classification_data
957 def _process_sfi_params(
958 target_sfi
: Dict
[str, Any
],
959 indata
: Dict
[str, Any
],
960 vim_info
: Dict
[str, Any
],
961 target_record_id
: str,
962 **kwargs
: Dict
[str, Any
],
967 target_sfi (Dict[str, Any]): SFI dictionary parameters that needs to be processed to create resource on VIM
968 indata (Dict[str, Any]): deployment info
969 vim_info (Dict[str, Any]): To add items created by OSM on the VIM.
970 target_record_id (str): Task record ID.
971 **kwargs (Dict[str, Any]): Used to send additional information to the task.
974 Dict[str, Any]: Return parameters required to create SFI and Items on which SFI is dependent.
977 vnfr_id
= target_sfi
["vnfr_id"]
978 vdur_id
= target_sfi
["vdur_id"]
981 "name": target_sfi
["id"],
982 "ingress_port_index": target_sfi
["ingress_port_index"],
983 "egress_port_index": target_sfi
["egress_port_index"],
986 db
= kwargs
.get("db")
987 vdu_text
= Ns
._get
_vnfr
_vdur
_text
(db
, vnfr_id
, vdur_id
)
989 extra_dict
= {"depends_on": [vdu_text
]}
990 sfi_data
["ingress_port"] = "TASK-" + vdu_text
991 sfi_data
["egress_port"] = "TASK-" + vdu_text
993 extra_dict
["params"] = sfi_data
998 def _get_vnfr_vdur_text(db
, vnfr_id
, vdur_id
):
999 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
1000 db_vnfr
= db
.get_one("vnfrs", {"_id": vnfr_id
})
1006 vdur
["id"] for vdur
in db_vnfr
["vdur"] if vdur
["vdu-id-ref"] == vdur_id
1010 vdu_text
= vnf_preffix
+ ":vdur." + vdur_list
[0]
1015 def _process_sf_params(
1016 target_sf
: Dict
[str, Any
],
1017 indata
: Dict
[str, Any
],
1018 vim_info
: Dict
[str, Any
],
1019 target_record_id
: str,
1020 **kwargs
: Dict
[str, Any
],
1021 ) -> Dict
[str, Any
]:
1025 target_sf (Dict[str, Any]): SF dictionary parameters that needs to be processed to create resource on VIM
1026 indata (Dict[str, Any]): Deployment info.
1027 vim_info (Dict[str, Any]):To add items created by OSM on the VIM.
1028 target_record_id (str): Task record ID.
1029 **kwargs (Dict[str, Any]): Used to send additional information to the task.
1032 Dict[str, Any]: Return parameters required to create SF and Items on which SF is dependent.
1035 nsr_id
= kwargs
.get("nsr_id", "")
1036 sfis
= target_sf
["sfis"]
1037 ns_preffix
= "nsrs:{}".format(nsr_id
)
1038 extra_dict
= {"depends_on": [], "params": []}
1039 sf_data
= {"name": target_sf
["id"], "sfis": sfis
}
1041 for count
, sfi
in enumerate(sfis
):
1042 sfi_text
= ns_preffix
+ ":sfi." + sfi
1043 sfis
[count
] = "TASK-" + sfi_text
1044 extra_dict
["depends_on"].append(sfi_text
)
1046 extra_dict
["params"] = sf_data
1051 def _process_sfp_params(
1052 target_sfp
: Dict
[str, Any
],
1053 indata
: Dict
[str, Any
],
1054 vim_info
: Dict
[str, Any
],
1055 target_record_id
: str,
1056 **kwargs
: Dict
[str, Any
],
1057 ) -> Dict
[str, Any
]:
1061 target_sfp (Dict[str, Any]): SFP dictionary parameters that needs to be processed to create resource on VIM.
1062 indata (Dict[str, Any]): Deployment info
1063 vim_info (Dict[str, Any]):To add items created by OSM on the VIM.
1064 target_record_id (str): Task record ID.
1065 **kwargs (Dict[str, Any]): Used to send additional information to the task.
1068 Dict[str, Any]: Return parameters required to create SFP and Items on which SFP is dependent.
1071 nsr_id
= kwargs
.get("nsr_id")
1072 sfs
= target_sfp
["sfs"]
1073 classifications
= target_sfp
["classifications"]
1074 ns_preffix
= "nsrs:{}".format(nsr_id
)
1075 extra_dict
= {"depends_on": [], "params": []}
1077 "name": target_sfp
["id"],
1079 "classifications": classifications
,
1082 for count
, sf
in enumerate(sfs
):
1083 sf_text
= ns_preffix
+ ":sf." + sf
1084 sfs
[count
] = "TASK-" + sf_text
1085 extra_dict
["depends_on"].append(sf_text
)
1087 for count
, classi
in enumerate(classifications
):
1088 classi_text
= ns_preffix
+ ":classification." + classi
1089 classifications
[count
] = "TASK-" + classi_text
1090 extra_dict
["depends_on"].append(classi_text
)
1092 extra_dict
["params"] = sfp_data
1097 def _process_net_params(
1098 target_vld
: Dict
[str, Any
],
1099 indata
: Dict
[str, Any
],
1100 vim_info
: Dict
[str, Any
],
1101 target_record_id
: str,
1102 **kwargs
: Dict
[str, Any
],
1103 ) -> Dict
[str, Any
]:
1104 """Function to process network parameters.
1107 target_vld (Dict[str, Any]): [description]
1108 indata (Dict[str, Any]): [description]
1109 vim_info (Dict[str, Any]): [description]
1110 target_record_id (str): [description]
1113 Dict[str, Any]: [description]
1117 if vim_info
.get("sdn"):
1118 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
1119 # ns_preffix = "nsrs:{}".format(nsr_id)
1120 # remove the ending ".sdn
1121 vld_target_record_id
, _
, _
= target_record_id
.rpartition(".")
1122 extra_dict
["params"] = {
1124 for k
in ("sdn-ports", "target_vim", "vlds", "type")
1128 # TODO needed to add target_id in the dependency.
1129 if vim_info
.get("target_vim"):
1130 extra_dict
["depends_on"] = [
1131 f
"{vim_info.get('target_vim')} {vld_target_record_id}"
1136 if vim_info
.get("vim_network_name"):
1137 extra_dict
["find_params"] = {
1139 "name": vim_info
.get("vim_network_name"),
1142 elif vim_info
.get("vim_network_id"):
1143 extra_dict
["find_params"] = {
1145 "id": vim_info
.get("vim_network_id"),
1148 elif target_vld
.get("mgmt-network") and not vim_info
.get("provider_network"):
1149 extra_dict
["find_params"] = {
1151 "name": target_vld
["id"],
1155 extra_dict
["params"] = {
1157 f
"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
1159 "ip_profile": vim_info
.get("ip_profile"),
1160 "provider_network_profile": vim_info
.get("provider_network"),
1163 if not target_vld
.get("underlay"):
1164 extra_dict
["params"]["net_type"] = "bridge"
1166 extra_dict
["params"]["net_type"] = (
1167 "ptp" if target_vld
.get("type") == "ELINE" else "data"
1173 def find_persistent_root_volumes(
1176 vdu_instantiation_volumes_list
: list,
1178 ) -> Dict
[str, any
]:
1179 """Find the persistent root volumes and add them to the disk_list
1180 by parsing the instantiation parameters.
1183 vnfd (dict): VNF descriptor
1184 target_vdu (dict): processed VDU
1185 vdu_instantiation_volumes_list (list): instantiation parameters for the each VDU as a list
1186 disk_list (list): to be filled up
1189 persistent_root_disk (dict): Details of persistent root disk
1192 persistent_root_disk
= {}
1193 # There can be only one root disk, when we find it, it will return the result
1195 for vdu
, vsd
in product(
1196 vnfd
.get("vdu", ()), vnfd
.get("virtual-storage-desc", ())
1199 vdu
["name"] == target_vdu
["vdu-name"]
1200 and vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]
1204 root_disk
.get("type-of-storage")
1205 == "persistent-storage:persistent-storage"
1207 for vdu_volume
in vdu_instantiation_volumes_list
:
1209 vdu_volume
["vim-volume-id"]
1210 and root_disk
["id"] == vdu_volume
["name"]
1212 persistent_root_disk
[vsd
["id"]] = {
1213 "vim_volume_id": vdu_volume
["vim-volume-id"],
1214 "image_id": vdu
.get("sw-image-desc"),
1217 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1219 return persistent_root_disk
1222 if root_disk
.get("size-of-storage"):
1223 persistent_root_disk
[vsd
["id"]] = {
1224 "image_id": vdu
.get("sw-image-desc"),
1225 "size": root_disk
.get("size-of-storage"),
1226 "keep": Ns
.is_volume_keeping_required(root_disk
),
1229 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1231 return persistent_root_disk
1232 return persistent_root_disk
1235 def find_persistent_volumes(
1236 persistent_root_disk
: dict,
1238 vdu_instantiation_volumes_list
: list,
1241 """Find the ordinary persistent volumes and add them to the disk_list
1242 by parsing the instantiation parameters.
1245 persistent_root_disk: persistent root disk dictionary
1246 target_vdu: processed VDU
1247 vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list
1248 disk_list: to be filled up
1251 # Find the ordinary volumes which are not added to the persistent_root_disk
1252 persistent_disk
= {}
1253 for disk
in target_vdu
.get("virtual-storages", {}):
1255 disk
.get("type-of-storage") == "persistent-storage:persistent-storage"
1256 and disk
["id"] not in persistent_root_disk
.keys()
1258 for vdu_volume
in vdu_instantiation_volumes_list
:
1259 if vdu_volume
["vim-volume-id"] and disk
["id"] == vdu_volume
["name"]:
1260 persistent_disk
[disk
["id"]] = {
1261 "vim_volume_id": vdu_volume
["vim-volume-id"],
1263 disk_list
.append(persistent_disk
[disk
["id"]])
1266 if disk
["id"] not in persistent_disk
.keys():
1267 persistent_disk
[disk
["id"]] = {
1268 "size": disk
.get("size-of-storage"),
1269 "keep": Ns
.is_volume_keeping_required(disk
),
1271 disk_list
.append(persistent_disk
[disk
["id"]])
1274 def is_volume_keeping_required(virtual_storage_desc
: Dict
[str, Any
]) -> bool:
1275 """Function to decide keeping persistent volume
1279 virtual_storage_desc (Dict[str, Any]): virtual storage description dictionary
1285 if not virtual_storage_desc
.get("vdu-storage-requirements"):
1287 for item
in virtual_storage_desc
.get("vdu-storage-requirements", {}):
1288 if item
.get("key") == "keep-volume" and item
.get("value").lower() == "true":
1293 def is_shared_volume(
1294 virtual_storage_desc
: Dict
[str, Any
], vnfd_id
: str
1296 """Function to decide if the volume type is multi attached or not .
1299 virtual_storage_desc (Dict[str, Any]): virtual storage description dictionary
1300 vnfd_id (str): vnfd id
1304 name (str) New name if it is a multiattach disk
1307 if vdu_storage_requirements
:= virtual_storage_desc
.get(
1308 "vdu-storage-requirements", {}
1310 for item
in vdu_storage_requirements
:
1312 item
.get("key") == "multiattach"
1313 and item
.get("value").lower() == "true"
1315 name
= f
"shared-{virtual_storage_desc['id']}-{vnfd_id}"
1317 return virtual_storage_desc
["id"], False
1320 def _sort_vdu_interfaces(target_vdu
: dict) -> None:
1321 """Sort the interfaces according to position number.
1324 target_vdu (dict): Details of VDU to be created
1327 # If the position info is provided for all the interfaces, it will be sorted
1328 # according to position number ascendingly.
1329 sorted_interfaces
= sorted(
1330 target_vdu
["interfaces"],
1331 key
=lambda x
: (x
.get("position") is None, x
.get("position")),
1333 target_vdu
["interfaces"] = sorted_interfaces
1336 def _partially_locate_vdu_interfaces(target_vdu
: dict) -> None:
1337 """Only place the interfaces which has specific position.
1340 target_vdu (dict): Details of VDU to be created
1343 # If the position info is provided for some interfaces but not all of them, the interfaces
1344 # which has specific position numbers will be placed and others' positions will not be taken care.
1346 i
.get("position") + 1
1347 for i
in target_vdu
["interfaces"]
1348 if i
.get("position") is not None
1350 n
= len(target_vdu
["interfaces"])
1351 sorted_interfaces
= [-1] * n
1355 if target_vdu
["interfaces"][k
].get("position") is not None:
1356 if any(i
.get("position") == 0 for i
in target_vdu
["interfaces"]):
1357 idx
= target_vdu
["interfaces"][k
]["position"] + 1
1359 idx
= target_vdu
["interfaces"][k
]["position"]
1360 sorted_interfaces
[idx
- 1] = target_vdu
["interfaces"][k
]
1364 if target_vdu
["interfaces"][m
].get("position") is None:
1365 idy
= sorted_interfaces
.index(-1)
1366 sorted_interfaces
[idy
] = target_vdu
["interfaces"][m
]
1369 target_vdu
["interfaces"] = sorted_interfaces
1372 def _prepare_vdu_cloud_init(
1373 target_vdu
: dict, vdu2cloud_init
: dict, db
: object, fs
: object
1375 """Fill cloud_config dict with cloud init details.
1378 target_vdu (dict): Details of VDU to be created
1379 vdu2cloud_init (dict): Cloud init dict
1380 db (object): DB object
1381 fs (object): FS object
1384 cloud_config (dict): Cloud config details of VDU
1390 if target_vdu
.get("cloud-init"):
1391 if target_vdu
["cloud-init"] not in vdu2cloud_init
:
1392 vdu2cloud_init
[target_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1395 location
=target_vdu
["cloud-init"],
1398 cloud_content_
= vdu2cloud_init
[target_vdu
["cloud-init"]]
1399 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1400 cloud_init_content
=cloud_content_
,
1401 params
=target_vdu
.get("additionalParams"),
1402 context
=target_vdu
["cloud-init"],
1405 if target_vdu
.get("boot-data-drive"):
1406 cloud_config
["boot-data-drive"] = target_vdu
.get("boot-data-drive")
1411 def _check_vld_information_of_interfaces(
1412 interface
: dict, ns_preffix
: str, vnf_preffix
: str
1414 """Prepare the net_text by the virtual link information for vnf and ns level.
1416 interface (dict): Interface details
1417 ns_preffix (str): Prefix of NS
1418 vnf_preffix (str): Prefix of VNF
1421 net_text (str): information of net
1425 if interface
.get("ns-vld-id"):
1426 net_text
= ns_preffix
+ ":vld." + interface
["ns-vld-id"]
1427 elif interface
.get("vnf-vld-id"):
1428 net_text
= vnf_preffix
+ ":vld." + interface
["vnf-vld-id"]
1433 def _prepare_interface_port_security(interface
: dict) -> None:
1437 interface (dict): Interface details
1440 if "port-security-enabled" in interface
:
1441 interface
["port_security"] = interface
.pop("port-security-enabled")
1443 if "port-security-disable-strategy" in interface
:
1444 interface
["port_security_disable_strategy"] = interface
.pop(
1445 "port-security-disable-strategy"
1449 def _create_net_item_of_interface(interface
: dict, net_text
: str) -> dict:
1450 """Prepare net item including name, port security, floating ip etc.
1453 interface (dict): Interface details
1454 net_text (str): information of net
1457 net_item (dict): Dict including net details
1463 for x
, v
in interface
.items()
1469 "port_security_disable_strategy",
1473 net_item
["net_id"] = "TASK-" + net_text
1474 net_item
["type"] = "virtual"
1479 def _prepare_type_of_interface(
1480 interface
: dict, tasks_by_target_record_id
: dict, net_text
: str, net_item
: dict
1482 """Fill the net item type by interface type such as SR-IOV, OM-MGMT, bridge etc.
1485 interface (dict): Interface details
1486 tasks_by_target_record_id (dict): Task details
1487 net_text (str): information of net
1488 net_item (dict): Dict including net details
1491 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1492 # TODO floating_ip: True/False (or it can be None)
1494 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1495 # Mark the net create task as type data
1497 tasks_by_target_record_id
,
1503 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"][
1507 net_item
["use"] = "data"
1508 net_item
["model"] = interface
["type"]
1509 net_item
["type"] = interface
["type"]
1512 interface
.get("type") == "OM-MGMT"
1513 or interface
.get("mgmt-interface")
1514 or interface
.get("mgmt-vnf")
1516 net_item
["use"] = "mgmt"
1519 # If interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1520 net_item
["use"] = "bridge"
1521 net_item
["model"] = interface
.get("type")
1524 def _prepare_vdu_interfaces(
1530 tasks_by_target_record_id
: dict,
1533 """Prepare the net_item and add net_list, add mgmt interface to extra_dict.
1536 target_vdu (dict): VDU to be created
1537 extra_dict (dict): Dictionary to be filled
1538 ns_preffix (str): NS prefix as string
1539 vnf_preffix (str): VNF prefix as string
1540 logger (object): Logger Object
1541 tasks_by_target_record_id (dict): Task details
1542 net_list (list): Net list of VDU
1544 for iface_index
, interface
in enumerate(target_vdu
["interfaces"]):
1545 net_text
= Ns
._check
_vld
_information
_of
_interfaces
(
1546 interface
, ns_preffix
, vnf_preffix
1549 # Interface not connected to any vld
1551 "Interface {} from vdu {} not connected to any vld".format(
1552 iface_index
, target_vdu
["vdu-name"]
1557 extra_dict
["depends_on"].append(net_text
)
1559 Ns
._prepare
_interface
_port
_security
(interface
)
1561 net_item
= Ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
1563 Ns
._prepare
_type
_of
_interface
(
1564 interface
, tasks_by_target_record_id
, net_text
, net_item
1567 if interface
.get("ip-address"):
1568 net_item
["ip_address"] = interface
["ip-address"]
1570 if interface
.get("mac-address"):
1571 net_item
["mac_address"] = interface
["mac-address"]
1573 net_list
.append(net_item
)
1575 if interface
.get("mgmt-vnf"):
1576 extra_dict
["mgmt_vnf_interface"] = iface_index
1577 elif interface
.get("mgmt-interface"):
1578 extra_dict
["mgmt_vdu_interface"] = iface_index
1581 def _prepare_vdu_ssh_keys(
1582 target_vdu
: dict, ro_nsr_public_key
: dict, cloud_config
: dict
1584 """Add ssh keys to cloud config.
1587 target_vdu (dict): Details of VDU to be created
1588 ro_nsr_public_key (dict): RO NSR public Key
1589 cloud_config (dict): Cloud config details
1594 if target_vdu
.get("ssh-keys"):
1595 ssh_keys
+= target_vdu
.get("ssh-keys")
1597 if target_vdu
.get("ssh-access-required"):
1598 ssh_keys
.append(ro_nsr_public_key
)
1601 cloud_config
["key-pairs"] = ssh_keys
1604 def _select_persistent_root_disk(vsd
: dict, vdu
: dict) -> dict:
1605 """Selects the persistent root disk if exists.
1607 vsd (dict): Virtual storage descriptors in VNFD
1608 vdu (dict): VNF descriptor
1611 root_disk (dict): Selected persistent root disk
1613 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
1617 ) == "persistent-storage:persistent-storage" and root_disk
.get(
1623 def _add_persistent_root_disk_to_disk_list(
1624 vnfd
: dict, target_vdu
: dict, persistent_root_disk
: dict, disk_list
: list
1626 """Find the persistent root disk and add to disk list.
1629 vnfd (dict): VNF descriptor
1630 target_vdu (dict): Details of VDU to be created
1631 persistent_root_disk (dict): Details of persistent root disk
1632 disk_list (list): Disks of VDU
1635 for vdu
in vnfd
.get("vdu", ()):
1636 if vdu
["name"] == target_vdu
["vdu-name"]:
1637 for vsd
in vnfd
.get("virtual-storage-desc", ()):
1638 root_disk
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
1642 persistent_root_disk
[vsd
["id"]] = {
1643 "image_id": vdu
.get("sw-image-desc"),
1644 "size": root_disk
["size-of-storage"],
1645 "keep": Ns
.is_volume_keeping_required(root_disk
),
1647 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1651 def _add_persistent_ordinary_disks_to_disk_list(
1653 persistent_root_disk
: dict,
1654 persistent_ordinary_disk
: dict,
1660 """Fill the disk list by adding persistent ordinary disks.
1663 target_vdu (dict): Details of VDU to be created
1664 persistent_root_disk (dict): Details of persistent root disk
1665 persistent_ordinary_disk (dict): Details of persistent ordinary disk
1666 disk_list (list): Disks of VDU
1669 if target_vdu
.get("virtual-storages"):
1670 for disk
in target_vdu
["virtual-storages"]:
1672 disk
.get("type-of-storage")
1673 == "persistent-storage:persistent-storage"
1674 and disk
["id"] not in persistent_root_disk
.keys()
1676 name
, multiattach
= Ns
.is_shared_volume(disk
, vnf_id
)
1677 persistent_ordinary_disk
[disk
["id"]] = {
1679 "size": disk
["size-of-storage"],
1680 "keep": Ns
.is_volume_keeping_required(disk
),
1681 "multiattach": multiattach
,
1683 disk_list
.append(persistent_ordinary_disk
[disk
["id"]])
1684 if multiattach
: # VDU creation has to wait for shared volumes
1685 extra_dict
["depends_on"].append(
1686 f
"nsrs:{nsr_id}:shared-volumes.{name}"
1690 def _prepare_vdu_affinity_group_list(
1691 target_vdu
: dict, extra_dict
: dict, ns_preffix
: str
1692 ) -> List
[Dict
[str, any
]]:
1693 """Process affinity group details to prepare affinity group list.
1696 target_vdu (dict): Details of VDU to be created
1697 extra_dict (dict): Dictionary to be filled
1698 ns_preffix (str): Prefix as string
1702 affinity_group_list (list): Affinity group details
1705 affinity_group_list
= []
1707 if target_vdu
.get("affinity-or-anti-affinity-group-id"):
1708 for affinity_group_id
in target_vdu
["affinity-or-anti-affinity-group-id"]:
1710 affinity_group_text
= (
1711 ns_preffix
+ ":affinity-or-anti-affinity-group." + affinity_group_id
1714 if not isinstance(extra_dict
.get("depends_on"), list):
1715 raise NsException("Invalid extra_dict format.")
1717 extra_dict
["depends_on"].append(affinity_group_text
)
1718 affinity_group
["affinity_group_id"] = "TASK-" + affinity_group_text
1719 affinity_group_list
.append(affinity_group
)
1721 return affinity_group_list
1724 def _process_vdu_params(
1725 target_vdu
: Dict
[str, Any
],
1726 indata
: Dict
[str, Any
],
1727 vim_info
: Dict
[str, Any
],
1728 target_record_id
: str,
1729 **kwargs
: Dict
[str, Any
],
1730 ) -> Dict
[str, Any
]:
1731 """Function to process VDU parameters.
1734 target_vdu (Dict[str, Any]): [description]
1735 indata (Dict[str, Any]): [description]
1736 vim_info (Dict[str, Any]): [description]
1737 target_record_id (str): [description]
1740 Dict[str, Any]: [description]
1742 vnfr_id
= kwargs
.get("vnfr_id")
1743 nsr_id
= kwargs
.get("nsr_id")
1744 vnfr
= kwargs
.get("vnfr")
1745 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1746 tasks_by_target_record_id
= kwargs
.get("tasks_by_target_record_id")
1747 logger
= kwargs
.get("logger")
1748 db
= kwargs
.get("db")
1749 fs
= kwargs
.get("fs")
1750 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1752 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
1753 ns_preffix
= "nsrs:{}".format(nsr_id
)
1754 image_text
= ns_preffix
+ ":image." + target_vdu
["ns-image-id"]
1755 flavor_text
= ns_preffix
+ ":flavor." + target_vdu
["ns-flavor-id"]
1756 extra_dict
= {"depends_on": [image_text
, flavor_text
]}
1758 persistent_root_disk
= {}
1759 persistent_ordinary_disk
= {}
1760 vdu_instantiation_volumes_list
= []
1762 vnfd_id
= vnfr
["vnfd-id"]
1763 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
1764 # If the position info is provided for all the interfaces, it will be sorted
1765 # according to position number ascendingly.
1767 True if i
.get("position") is not None else False
1768 for i
in target_vdu
["interfaces"]
1770 Ns
._sort
_vdu
_interfaces
(target_vdu
)
1772 # If the position info is provided for some interfaces but not all of them, the interfaces
1773 # which has specific position numbers will be placed and others' positions will not be taken care.
1775 Ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
1777 # If the position info is not provided for the interfaces, interfaces will be attached
1778 # according to the order in the VNFD.
1779 Ns
._prepare
_vdu
_interfaces
(
1785 tasks_by_target_record_id
,
1790 cloud_config
= Ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
1792 # Prepare VDU ssh keys
1793 Ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
1795 if target_vdu
.get("additionalParams"):
1796 vdu_instantiation_volumes_list
= (
1797 target_vdu
.get("additionalParams").get("OSM", {}).get("vdu_volumes")
1800 if vdu_instantiation_volumes_list
:
1801 # Find the root volumes and add to the disk_list
1802 persistent_root_disk
= Ns
.find_persistent_root_volumes(
1803 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
1806 # Find the ordinary volumes which are not added to the persistent_root_disk
1807 # and put them to the disk list
1808 Ns
.find_persistent_volumes(
1809 persistent_root_disk
,
1811 vdu_instantiation_volumes_list
,
1816 # Vdu_instantiation_volumes_list is empty
1817 # First get add the persistent root disks to disk_list
1818 Ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
1819 vnfd
, target_vdu
, persistent_root_disk
, disk_list
1821 # Add the persistent non-root disks to disk_list
1822 Ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
1824 persistent_root_disk
,
1825 persistent_ordinary_disk
,
1832 affinity_group_list
= Ns
._prepare
_vdu
_affinity
_group
_list
(
1833 target_vdu
, extra_dict
, ns_preffix
1836 extra_dict
["params"] = {
1837 "name": "{}-{}-{}-{}".format(
1838 indata
["name"][:16],
1839 vnfr
["member-vnf-index-ref"][:16],
1840 target_vdu
["vdu-name"][:32],
1841 target_vdu
.get("count-index") or 0,
1843 "description": target_vdu
["vdu-name"],
1845 "image_id": "TASK-" + image_text
,
1846 "flavor_id": "TASK-" + flavor_text
,
1847 "affinity_group_list": affinity_group_list
,
1848 "net_list": net_list
,
1849 "cloud_config": cloud_config
or None,
1850 "disk_list": disk_list
,
1851 "availability_zone_index": None, # TODO
1852 "availability_zone_list": None, # TODO
1857 def _process_shared_volumes_params(
1858 target_shared_volume
: Dict
[str, Any
],
1859 indata
: Dict
[str, Any
],
1860 vim_info
: Dict
[str, Any
],
1861 target_record_id
: str,
1862 **kwargs
: Dict
[str, Any
],
1863 ) -> Dict
[str, Any
]:
1865 shared_volume_data
= {
1866 "size": target_shared_volume
["size-of-storage"],
1867 "name": target_shared_volume
["id"],
1868 "type": target_shared_volume
["type-of-storage"],
1869 "keep": Ns
.is_volume_keeping_required(target_shared_volume
),
1871 extra_dict
["params"] = shared_volume_data
1875 def _process_affinity_group_params(
1876 target_affinity_group
: Dict
[str, Any
],
1877 indata
: Dict
[str, Any
],
1878 vim_info
: Dict
[str, Any
],
1879 target_record_id
: str,
1880 **kwargs
: Dict
[str, Any
],
1881 ) -> Dict
[str, Any
]:
1882 """Get affinity or anti-affinity group parameters.
1885 target_affinity_group (Dict[str, Any]): [description]
1886 indata (Dict[str, Any]): [description]
1887 vim_info (Dict[str, Any]): [description]
1888 target_record_id (str): [description]
1891 Dict[str, Any]: [description]
1895 affinity_group_data
= {
1896 "name": target_affinity_group
["name"],
1897 "type": target_affinity_group
["type"],
1898 "scope": target_affinity_group
["scope"],
1901 if target_affinity_group
.get("vim-affinity-group-id"):
1902 affinity_group_data
["vim-affinity-group-id"] = target_affinity_group
[
1903 "vim-affinity-group-id"
1906 extra_dict
["params"] = {
1907 "affinity_group_data": affinity_group_data
,
1912 def _process_recreate_vdu_params(
1913 existing_vdu
: Dict
[str, Any
],
1914 db_nsr
: Dict
[str, Any
],
1915 vim_info
: Dict
[str, Any
],
1916 target_record_id
: str,
1918 **kwargs
: Dict
[str, Any
],
1919 ) -> Dict
[str, Any
]:
1920 """Function to process VDU parameters to recreate.
1923 existing_vdu (Dict[str, Any]): [description]
1924 db_nsr (Dict[str, Any]): [description]
1925 vim_info (Dict[str, Any]): [description]
1926 target_record_id (str): [description]
1927 target_id (str): [description]
1930 Dict[str, Any]: [description]
1932 vnfr
= kwargs
.get("vnfr")
1933 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1934 # logger = kwargs.get("logger")
1935 db
= kwargs
.get("db")
1936 fs
= kwargs
.get("fs")
1937 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1943 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
1945 if vim_details_text
:
1946 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
1948 for iface_index
, interface
in enumerate(existing_vdu
["interfaces"]):
1949 if "port-security-enabled" in interface
:
1950 interface
["port_security"] = interface
.pop("port-security-enabled")
1952 if "port-security-disable-strategy" in interface
:
1953 interface
["port_security_disable_strategy"] = interface
.pop(
1954 "port-security-disable-strategy"
1959 for x
, v
in interface
.items()
1965 "port_security_disable_strategy",
1969 existing_ifaces
= existing_vdu
["vim_info"][target_id
].get(
1970 "interfaces_backup", []
1975 for i
in existing_ifaces
1976 if i
["ip_address"] == interface
["ip-address"]
1981 net_item
["net_id"] = net_id
1982 net_item
["type"] = "virtual"
1984 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1985 # TODO floating_ip: True/False (or it can be None)
1986 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1987 net_item
["use"] = "data"
1988 net_item
["model"] = interface
["type"]
1989 net_item
["type"] = interface
["type"]
1991 interface
.get("type") == "OM-MGMT"
1992 or interface
.get("mgmt-interface")
1993 or interface
.get("mgmt-vnf")
1995 net_item
["use"] = "mgmt"
1997 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1998 net_item
["use"] = "bridge"
1999 net_item
["model"] = interface
.get("type")
2001 if interface
.get("ip-address"):
2002 net_item
["ip_address"] = interface
["ip-address"]
2004 if interface
.get("mac-address"):
2005 net_item
["mac_address"] = interface
["mac-address"]
2007 net_list
.append(net_item
)
2009 if interface
.get("mgmt-vnf"):
2010 extra_dict
["mgmt_vnf_interface"] = iface_index
2011 elif interface
.get("mgmt-interface"):
2012 extra_dict
["mgmt_vdu_interface"] = iface_index
2017 if existing_vdu
.get("cloud-init"):
2018 if existing_vdu
["cloud-init"] not in vdu2cloud_init
:
2019 vdu2cloud_init
[existing_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
2022 location
=existing_vdu
["cloud-init"],
2025 cloud_content_
= vdu2cloud_init
[existing_vdu
["cloud-init"]]
2026 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
2027 cloud_init_content
=cloud_content_
,
2028 params
=existing_vdu
.get("additionalParams"),
2029 context
=existing_vdu
["cloud-init"],
2032 if existing_vdu
.get("boot-data-drive"):
2033 cloud_config
["boot-data-drive"] = existing_vdu
.get("boot-data-drive")
2037 if existing_vdu
.get("ssh-keys"):
2038 ssh_keys
+= existing_vdu
.get("ssh-keys")
2040 if existing_vdu
.get("ssh-access-required"):
2041 ssh_keys
.append(ro_nsr_public_key
)
2044 cloud_config
["key-pairs"] = ssh_keys
2047 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
2048 disk_list
.append({"vim_id": vol_id
["id"]})
2050 affinity_group_list
= []
2052 if existing_vdu
.get("affinity-or-anti-affinity-group-id"):
2054 for affinity_group_id
in existing_vdu
["affinity-or-anti-affinity-group-id"]:
2055 for group
in db_nsr
.get("affinity-or-anti-affinity-group"):
2057 group
["id"] == affinity_group_id
2058 and group
["vim_info"][target_id
].get("vim_id", None) is not None
2060 affinity_group
["affinity_group_id"] = group
["vim_info"][
2062 ].get("vim_id", None)
2063 affinity_group_list
.append(affinity_group
)
2065 extra_dict
["params"] = {
2066 "name": "{}-{}-{}-{}".format(
2067 db_nsr
["name"][:16],
2068 vnfr
["member-vnf-index-ref"][:16],
2069 existing_vdu
["vdu-name"][:32],
2070 existing_vdu
.get("count-index") or 0,
2072 "description": existing_vdu
["vdu-name"],
2074 "image_id": vim_details
["image"]["id"],
2075 "flavor_id": vim_details
["flavor"]["id"],
2076 "affinity_group_list": affinity_group_list
,
2077 "net_list": net_list
,
2078 "cloud_config": cloud_config
or None,
2079 "disk_list": disk_list
,
2080 "availability_zone_index": None, # TODO
2081 "availability_zone_list": None, # TODO
2086 def calculate_diff_items(
2093 tasks_by_target_record_id
,
2100 """Function that returns the incremental changes (creation, deletion)
2101 related to a specific item `item` to be done. This function should be
2102 called for NS instantiation, NS termination, NS update to add a new VNF
2103 or a new VLD, remove a VNF or VLD, etc.
2104 Item can be `net`, `flavor`, `image` or `vdu`.
2105 It takes a list of target items from indata (which came from the REST API)
2106 and compares with the existing items from db_ro_nsr, identifying the
2107 incremental changes to be done. During the comparison, it calls the method
2108 `process_params` (which was passed as parameter, and is particular for each
2112 indata (Dict[str, Any]): deployment info
2113 db_nsr: NSR record from DB
2114 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2115 db_nsr_update (Dict[str, Any]): NSR info to update in DB
2116 item (str): element to process (net, vdu...)
2117 tasks_by_target_record_id (Dict[str, Any]):
2118 [<target_record_id>, <task>]
2119 action_id (str): action id
2120 nsr_id (str): NSR id
2121 task_index (number): task index to add to task name
2122 vnfr_id (str): VNFR id
2123 vnfr (Dict[str, Any]): VNFR info
2126 List: list with the incremental changes (deletes, creates) for each item
2127 number: current task index
2135 process_params
= None
2136 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
2137 ro_nsr_public_key
= db_ro_nsr
["public_key"]
2138 # According to the type of item, the path, the target_list,
2139 # the existing_list and the method to process params are set
2140 db_path
= self
.db_path_map
[item
]
2141 process_params
= self
.process_params_function_map
[item
]
2143 if item
in ("sfp", "classification", "sf", "sfi"):
2144 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
2145 target_vnffg
= indata
.get("vnffg", [])[0]
2146 target_list
= target_vnffg
[item
]
2147 existing_list
= db_nsr
.get(item
, [])
2148 elif item
in ("net", "vdu"):
2149 # This case is specific for the NS VLD (not applied to VDU)
2151 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
2152 target_list
= indata
.get("ns", []).get(db_path
, [])
2153 existing_list
= db_nsr
.get(db_path
, [])
2154 # This case is common for VNF VLDs and VNF VDUs
2156 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
2158 (vnf
for vnf
in indata
.get("vnf", ()) if vnf
["_id"] == vnfr_id
),
2161 target_list
= target_vnf
.get(db_path
, []) if target_vnf
else []
2162 existing_list
= vnfr
.get(db_path
, [])
2166 "affinity-or-anti-affinity-group",
2169 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
2170 target_list
= indata
.get(item
, [])
2171 existing_list
= db_nsr
.get(item
, [])
2173 raise NsException("Item not supported: {}", item
)
2174 # ensure all the target_list elements has an "id". If not assign the index as id
2175 if target_list
is None:
2177 for target_index
, tl
in enumerate(target_list
):
2178 if tl
and not tl
.get("id"):
2179 tl
["id"] = str(target_index
)
2180 # step 1 items (networks,vdus,...) to be deleted/updated
2181 for item_index
, existing_item
in enumerate(existing_list
):
2183 (t
for t
in target_list
if t
["id"] == existing_item
["id"]),
2186 for target_vim
, existing_viminfo
in existing_item
.get(
2189 if existing_viminfo
is None:
2193 target_viminfo
= target_item
.get("vim_info", {}).get(target_vim
)
2195 target_viminfo
= None
2197 if target_viminfo
is None:
2199 self
._assign
_vim
(target_vim
)
2200 target_record_id
= "{}.{}".format(db_record
, existing_item
["id"])
2203 if target_vim
.startswith("sdn") or target_vim
.startswith("wim"):
2204 # item must be sdn-net instead of net if target_vim is a sdn
2206 target_record_id
+= ".sdn"
2209 "action_id": action_id
,
2211 "task_index": task_index
,
2216 "deployment_info": deployment_info
,
2217 "target_id": target_vim
,
2220 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
2221 "target_record_id": target_record_id
,
2226 # step 2 items (networks,vdus,...) to be created
2227 for target_item
in target_list
:
2229 for item_index
, existing_item
in enumerate(existing_list
):
2230 if existing_item
["id"] == target_item
["id"]:
2234 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
2235 existing_list
.append(target_item
)
2236 existing_item
= None
2238 for target_vim
, target_viminfo
in target_item
.get("vim_info", {}).items():
2239 existing_viminfo
= None
2242 existing_viminfo
= existing_item
.get("vim_info", {}).get(target_vim
)
2244 if existing_viminfo
is not None:
2247 target_record_id
= "{}.{}".format(db_record
, target_item
["id"])
2250 if target_vim
.startswith("sdn") or target_vim
.startswith("wim"):
2251 # item must be sdn-net instead of net if target_vim is a sdn
2253 target_record_id
+= ".sdn"
2257 "ns.calculate_diff_items target_item={}".format(target_item
)
2259 if process_params
== Ns
._process
_flavor
_params
:
2266 "calculate_diff_items for flavor kwargs={}".format(kwargs
)
2269 if process_params
== Ns
._process
_vdu
_params
:
2270 self
.logger
.debug("calculate_diff_items self.fs={}".format(self
.fs
))
2276 "vdu2cloud_init": vdu2cloud_init
,
2277 "tasks_by_target_record_id": tasks_by_target_record_id
,
2278 "logger": self
.logger
,
2281 "ro_nsr_public_key": ro_nsr_public_key
,
2284 self
.logger
.debug("calculate_diff_items kwargs={}".format(kwargs
))
2286 process_params
== Ns
._process
_sfi
_params
2287 or Ns
._process
_sf
_params
2288 or Ns
._process
_classification
_params
2289 or Ns
._process
_sfp
_params
2291 kwargs
.update({"nsr_id": nsr_id
, "db": self
.db
})
2293 self
.logger
.debug("calculate_diff_items kwargs={}".format(kwargs
))
2295 extra_dict
= process_params(
2302 self
._assign
_vim
(target_vim
)
2305 "action_id": action_id
,
2307 "task_index": task_index
,
2311 "deployment_info": deployment_info
,
2312 "target_id": target_vim
,
2315 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
2316 "target_record_id": target_record_id
,
2317 "extra_dict": extra_dict
,
2318 "common_id": target_item
.get("common_id", None),
2320 diff_items
.append(new_item
)
2321 tasks_by_target_record_id
[target_record_id
] = new_item
2324 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
2326 return diff_items
, task_index
2328 def _process_vnfgd_sfp(self
, sfp
):
2330 # getting sfp name, sfs and classifications in sfp to store it in processed_sfp
2331 processed_sfp
["id"] = sfp
["id"]
2333 sf
["id"] for sf
in sfp
.get("position-desc-id", [])[0].get("cp-profile-id")
2335 classifications_in_sfp
= [
2337 for classi
in sfp
.get("position-desc-id", [])[0].get("match-attributes")
2340 # creating a list of sfp with sfs and classifications
2341 processed_sfp
["sfs"] = sfs_in_sfp
2342 processed_sfp
["classifications"] = classifications_in_sfp
2344 return processed_sfp
2346 def _process_vnfgd_sf(self
, sf
):
2348 # getting name of sf
2349 processed_sf
["id"] = sf
["id"]
2350 # getting sfis in sf
2351 sfis_in_sf
= sf
.get("constituent-profile-elements")
2352 sorted_sfis
= sorted(sfis_in_sf
, key
=lambda i
: i
["order"])
2353 # getting sfis names
2354 processed_sf
["sfis"] = [sfi
["id"] for sfi
in sorted_sfis
]
2358 def _process_vnfgd_sfi(self
, sfi
, db_vnfrs
):
2360 # getting name of sfi
2361 processed_sfi
["id"] = sfi
["id"]
2363 # getting ports in sfi
2364 ingress_port
= sfi
["ingress-constituent-cpd-id"]
2365 egress_port
= sfi
["egress-constituent-cpd-id"]
2366 sfi_vnf_member_index
= sfi
["constituent-base-element-id"]
2368 processed_sfi
["ingress_port"] = ingress_port
2369 processed_sfi
["egress_port"] = egress_port
2371 all_vnfrs
= db_vnfrs
.values()
2375 for element
in all_vnfrs
2376 if element
["member-vnf-index-ref"] == sfi_vnf_member_index
2378 processed_sfi
["vnfr_id"] = sfi_vnfr
[0]["id"]
2380 sfi_vnfr_cp
= sfi_vnfr
[0]["connection-point"]
2382 ingress_port_index
= [
2383 c
for c
, element
in enumerate(sfi_vnfr_cp
) if element
["id"] == ingress_port
2385 ingress_port_index
= ingress_port_index
[0]
2387 processed_sfi
["vdur_id"] = sfi_vnfr_cp
[ingress_port_index
][
2388 "connection-point-vdu-id"
2390 processed_sfi
["ingress_port_index"] = ingress_port_index
2391 processed_sfi
["egress_port_index"] = ingress_port_index
2393 if egress_port
!= ingress_port
:
2394 egress_port_index
= [
2396 for c
, element
in enumerate(sfi_vnfr_cp
)
2397 if element
["id"] == egress_port
2399 processed_sfi
["egress_port_index"] = egress_port_index
2401 return processed_sfi
2403 def _process_vnfgd_classification(self
, classification
, db_vnfrs
):
2404 processed_classification
= {}
2406 processed_classification
= deepcopy(classification
)
2407 classi_vnf_member_index
= processed_classification
[
2408 "constituent-base-element-id"
2410 logical_source_port
= processed_classification
["constituent-cpd-id"]
2412 all_vnfrs
= db_vnfrs
.values()
2416 for element
in all_vnfrs
2417 if element
["member-vnf-index-ref"] == classi_vnf_member_index
2419 processed_classification
["vnfr_id"] = classi_vnfr
[0]["id"]
2421 classi_vnfr_cp
= classi_vnfr
[0]["connection-point"]
2423 ingress_port_index
= [
2425 for c
, element
in enumerate(classi_vnfr_cp
)
2426 if element
["id"] == logical_source_port
2428 ingress_port_index
= ingress_port_index
[0]
2430 processed_classification
["ingress_port_index"] = ingress_port_index
2431 processed_classification
["vdur_id"] = classi_vnfr_cp
[ingress_port_index
][
2432 "connection-point-vdu-id"
2435 return processed_classification
2437 def _update_db_nsr_with_vnffg(self
, processed_vnffg
, vim_info
, nsr_id
):
2438 """This method used to add viminfo dict to sfi, sf sfp and classification in indata and count info in db_nsr.
2441 processed_vnffg (Dict[str, Any]): deployment info
2442 vim_info (Dict): dictionary to store VIM resource information
2443 nsr_id (str): NSR id
2451 nsr_classification
= {}
2452 db_nsr_vnffg
= deepcopy(processed_vnffg
)
2454 for count
, sfi
in enumerate(processed_vnffg
["sfi"]):
2455 sfi
["vim_info"] = vim_info
2456 sfi_count
= "sfi.{}".format(count
)
2457 nsr_sfi
[sfi_count
] = db_nsr_vnffg
["sfi"][count
]
2459 self
.db
.set_list("nsrs", {"_id": nsr_id
}, nsr_sfi
)
2461 for count
, sf
in enumerate(processed_vnffg
["sf"]):
2462 sf
["vim_info"] = vim_info
2463 sf_count
= "sf.{}".format(count
)
2464 nsr_sf
[sf_count
] = db_nsr_vnffg
["sf"][count
]
2466 self
.db
.set_list("nsrs", {"_id": nsr_id
}, nsr_sf
)
2468 for count
, sfp
in enumerate(processed_vnffg
["sfp"]):
2469 sfp
["vim_info"] = vim_info
2470 sfp_count
= "sfp.{}".format(count
)
2471 nsr_sfp
[sfp_count
] = db_nsr_vnffg
["sfp"][count
]
2473 self
.db
.set_list("nsrs", {"_id": nsr_id
}, nsr_sfp
)
2475 for count
, classi
in enumerate(processed_vnffg
["classification"]):
2476 classi
["vim_info"] = vim_info
2477 classification_count
= "classification.{}".format(count
)
2478 nsr_classification
[classification_count
] = db_nsr_vnffg
["classification"][
2482 self
.db
.set_list("nsrs", {"_id": nsr_id
}, nsr_classification
)
2484 def process_vnffgd_descriptor(
2491 """This method used to process vnffgd parameters from descriptor.
2494 indata (Dict[str, Any]): deployment info
2495 nsr_id (str): NSR id
2496 db_nsr: NSR record from DB
2497 db_vnfrs: VNFRS record from DB
2500 Dict: Processed vnffg parameters.
2503 processed_vnffg
= {}
2504 vnffgd
= db_nsr
.get("nsd", {}).get("vnffgd")
2505 vnf_list
= indata
.get("vnf", [])
2509 vim_text
= "vim:" + vnf_list
[0].get("vim-account-id", "")
2512 vim_info
[vim_text
] = {}
2514 processed_classifications
= []
2518 # setting up intial empty entries for vnffg items in mongodb.
2526 "classification": [],
2532 sfps
= vnffg
.get("nfpd")
2534 processed_sfp
= self
._process
_vnfgd
_sfp
(sfp
)
2535 # appending the list of processed sfps
2536 processed_sfps
.append(processed_sfp
)
2538 # getting sfs in sfp
2539 sfs
= sfp
.get("position-desc-id")[0].get("cp-profile-id")
2541 processed_sf
= self
._process
_vnfgd
_sf
(sf
)
2543 # appending the list of processed sfs
2544 processed_sfs
.append(processed_sf
)
2546 # getting sfis in sf
2547 sfis_in_sf
= sf
.get("constituent-profile-elements")
2548 sorted_sfis
= sorted(sfis_in_sf
, key
=lambda i
: i
["order"])
2550 for sfi
in sorted_sfis
:
2551 processed_sfi
= self
._process
_vnfgd
_sfi
(sfi
, db_vnfrs
)
2553 processed_sfis
.append(processed_sfi
)
2555 classifications
= sfp
.get("position-desc-id")[0].get("match-attributes")
2556 # getting classifications from sfp
2557 for classification
in classifications
:
2558 processed_classification
= self
._process
_vnfgd
_classification
(
2559 classification
, db_vnfrs
2562 processed_classifications
.append(processed_classification
)
2564 processed_vnffg
["sfi"] = processed_sfis
2565 processed_vnffg
["sf"] = processed_sfs
2566 processed_vnffg
["classification"] = processed_classifications
2567 processed_vnffg
["sfp"] = processed_sfps
2569 # adding viminfo dict to sfi, sf sfp and classification
2570 self
._update
_db
_nsr
_with
_vnffg
(processed_vnffg
, vim_info
, nsr_id
)
2572 # updating indata with vnffg porcessed parameters
2573 indata
["vnffg"].append(processed_vnffg
)
2575 def calculate_all_differences_to_deploy(
2585 tasks_by_target_record_id
,
2587 """This method calculates the ordered list of items (`changes_list`)
2588 to be created and deleted.
2591 indata (Dict[str, Any]): deployment info
2592 nsr_id (str): NSR id
2593 db_nsr: NSR record from DB
2594 db_vnfrs: VNFRS record from DB
2595 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2596 db_nsr_update (Dict[str, Any]): NSR info to update in DB
2597 db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
2598 action_id (str): action id
2599 tasks_by_target_record_id (Dict[str, Any]):
2600 [<target_record_id>, <task>]
2603 List: ordered list of items to be created and deleted.
2607 # set list with diffs:
2610 # processing vnffg from descriptor parameter
2611 vnffgd
= db_nsr
.get("nsd").get("vnffgd")
2612 if vnffgd
is not None:
2613 indata
["vnffg"] = []
2614 vnf_list
= indata
["vnf"]
2615 processed_vnffg
= {}
2617 # in case of ns-delete
2619 processed_vnffg
["sfi"] = []
2620 processed_vnffg
["sf"] = []
2621 processed_vnffg
["classification"] = []
2622 processed_vnffg
["sfp"] = []
2624 indata
["vnffg"].append(processed_vnffg
)
2627 self
.process_vnffgd_descriptor(
2634 # getting updated db_nsr having vnffg parameters
2635 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2638 "After processing vnffd parameters indata={} nsr={}".format(
2643 for item
in ["sfp", "classification", "sf", "sfi"]:
2644 self
.logger
.debug("process NS={} {}".format(nsr_id
, item
))
2645 diff_items
, task_index
= self
.calculate_diff_items(
2648 db_ro_nsr
=db_ro_nsr
,
2649 db_nsr_update
=db_nsr_update
,
2651 tasks_by_target_record_id
=tasks_by_target_record_id
,
2652 action_id
=action_id
,
2654 task_index
=task_index
,
2657 changes_list
+= diff_items
2659 # NS vld, image and flavor
2664 "affinity-or-anti-affinity-group",
2666 self
.logger
.debug("process NS={} {}".format(nsr_id
, item
))
2667 diff_items
, task_index
= self
.calculate_diff_items(
2670 db_ro_nsr
=db_ro_nsr
,
2671 db_nsr_update
=db_nsr_update
,
2673 tasks_by_target_record_id
=tasks_by_target_record_id
,
2674 action_id
=action_id
,
2676 task_index
=task_index
,
2679 changes_list
+= diff_items
2682 for vnfr_id
, vnfr
in db_vnfrs
.items():
2683 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
2684 for item
in ["net", "vdu", "shared-volumes"]:
2685 self
.logger
.debug("process VNF={} {}".format(vnfr_id
, item
))
2686 diff_items
, task_index
= self
.calculate_diff_items(
2689 db_ro_nsr
=db_ro_nsr
,
2690 db_nsr_update
=db_vnfrs_update
[vnfr
["_id"]],
2692 tasks_by_target_record_id
=tasks_by_target_record_id
,
2693 action_id
=action_id
,
2695 task_index
=task_index
,
2699 changes_list
+= diff_items
2703 def define_all_tasks(
2707 tasks_by_target_record_id
,
2709 """Function to create all the task structures obtanied from
2710 the method calculate_all_differences_to_deploy
2713 changes_list (List): ordered list of items to be created or deleted
2714 db_new_tasks (List): tasks list to be created
2715 action_id (str): action id
2716 tasks_by_target_record_id (Dict[str, Any]):
2717 [<target_record_id>, <task>]
2721 for change
in changes_list
:
2722 task
= Ns
._create
_task
(
2723 deployment_info
=change
["deployment_info"],
2724 target_id
=change
["target_id"],
2725 item
=change
["item"],
2726 action
=change
["action"],
2727 target_record
=change
["target_record"],
2728 target_record_id
=change
["target_record_id"],
2729 extra_dict
=change
.get("extra_dict", None),
2732 self
.logger
.debug("ns.define_all_tasks task={}".format(task
))
2733 tasks_by_target_record_id
[change
["target_record_id"]] = task
2734 db_new_tasks
.append(task
)
2736 if change
.get("common_id"):
2737 task
["common_id"] = change
["common_id"]
2739 def upload_all_tasks(
2744 """Function to save all tasks in the common DB
2747 db_new_tasks (List): tasks list to be created
2748 now (time): current time
2752 nb_ro_tasks
= 0 # for logging
2754 for db_task
in db_new_tasks
:
2755 target_id
= db_task
.pop("target_id")
2756 common_id
= db_task
.get("common_id")
2758 # Do not chek tasks with vim_status DELETED
2759 # because in manual heealing there are two tasks for the same vdur:
2760 # one with vim_status deleted and the other one with the actual VM status.
2766 "target_id": target_id
,
2767 "tasks.common_id": common_id
,
2768 "vim_info.vim_status.ne": "DELETED",
2770 update_dict
={"to_check_at": now
, "modified_at": now
},
2771 push
={"tasks": db_task
},
2772 fail_on_empty
=False,
2776 if not self
.db
.set_one(
2779 "target_id": target_id
,
2780 "tasks.target_record": db_task
["target_record"],
2781 "vim_info.vim_status.ne": "DELETED",
2783 update_dict
={"to_check_at": now
, "modified_at": now
},
2784 push
={"tasks": db_task
},
2785 fail_on_empty
=False,
2788 self
.logger
.debug("Updating database, Creating ro_tasks")
2789 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
2791 self
.db
.create("ro_tasks", db_ro_task
)
2794 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2795 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
2799 def upload_recreate_tasks(
2804 """Function to save recreate tasks in the common DB
2807 db_new_tasks (List): tasks list to be created
2808 now (time): current time
2812 nb_ro_tasks
= 0 # for logging
2814 for db_task
in db_new_tasks
:
2815 target_id
= db_task
.pop("target_id")
2816 self
.logger
.debug("target_id={} db_task={}".format(target_id
, db_task
))
2818 action
= db_task
.get("action", None)
2821 self
.logger
.debug("Updating database, Creating ro_tasks")
2822 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
2824 # If DELETE task: the associated created items should be removed
2825 # (except persistent volumes):
2826 if action
== "DELETE":
2827 db_ro_task
["vim_info"]["created"] = True
2828 db_ro_task
["vim_info"]["created_items"] = db_task
.get(
2831 db_ro_task
["vim_info"]["volumes_to_hold"] = db_task
.get(
2832 "volumes_to_hold", []
2834 db_ro_task
["vim_info"]["vim_id"] = db_task
.get("vim_id", None)
2837 self
.logger
.debug("upload_all_tasks db_ro_task={}".format(db_ro_task
))
2838 self
.db
.create("ro_tasks", db_ro_task
)
2841 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2842 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
2846 def _prepare_created_items_for_healing(
2852 # Get created_items from ro_task
2853 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2854 for ro_task
in ro_tasks
:
2855 for task
in ro_task
["tasks"]:
2857 task
["target_record"] == target_record
2858 and task
["action"] == "CREATE"
2859 and ro_task
["vim_info"]["created_items"]
2861 created_items
= ro_task
["vim_info"]["created_items"]
2864 return created_items
2866 def _prepare_persistent_volumes_for_healing(
2871 # The associated volumes of the VM shouldn't be removed
2874 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
2875 if vim_details_text
:
2876 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
2878 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
2879 volumes_list
.append(vol_id
["id"])
2883 def prepare_changes_to_recreate(
2891 tasks_by_target_record_id
,
2893 """This method will obtain an ordered list of items (`changes_list`)
2894 to be created and deleted to meet the recreate request.
2898 "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id
, indata
)
2902 # set list with diffs:
2904 db_path
= self
.db_path_map
["vdu"]
2905 target_list
= indata
.get("healVnfData", {})
2906 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
2907 ro_nsr_public_key
= db_ro_nsr
["public_key"]
2909 # Check each VNF of the target
2910 for target_vnf
in target_list
:
2911 # Find this VNF in the list from DB, raise exception if vnfInstanceId is not found
2912 vnfr_id
= target_vnf
["vnfInstanceId"]
2913 existing_vnf
= db_vnfrs
.get(vnfr_id
, {})
2914 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
2915 # vim_account_id = existing_vnf.get("vim-account-id", "")
2917 target_vdus
= target_vnf
.get("additionalParams", {}).get("vdu", [])
2918 # Check each VDU of this VNF
2920 # Create target_vdu_list from DB, if VDUs are not specified
2922 for existing_vdu
in existing_vnf
.get("vdur"):
2923 vdu_name
= existing_vdu
.get("vdu-name", None)
2924 vdu_index
= existing_vdu
.get("count-index", 0)
2925 vdu_to_be_healed
= {"vdu-id": vdu_name
, "count-index": vdu_index
}
2926 target_vdus
.append(vdu_to_be_healed
)
2927 for target_vdu
in target_vdus
:
2928 vdu_name
= target_vdu
.get("vdu-id", None)
2929 # For multi instance VDU count-index is mandatory
2930 # For single session VDU count-indes is 0
2931 count_index
= target_vdu
.get("count-index", 0)
2933 existing_instance
= {}
2935 for instance
in existing_vnf
.get("vdur", {}):
2937 instance
["vdu-name"] == vdu_name
2938 and instance
["count-index"] == count_index
2940 existing_instance
= instance
2945 target_record_id
= "{}.{}".format(db_record
, existing_instance
["id"])
2947 # The target VIM is the one already existing in DB to recreate
2948 for target_vim
, target_viminfo
in existing_instance
.get(
2951 # step 1 vdu to be deleted
2952 self
._assign
_vim
(target_vim
)
2954 "action_id": action_id
,
2956 "task_index": task_index
,
2959 target_record
= f
"{db_record}.{item_index}.vim_info.{target_vim}"
2960 created_items
= self
._prepare
_created
_items
_for
_healing
(
2961 nsr_id
, target_record
2964 volumes_to_hold
= self
._prepare
_persistent
_volumes
_for
_healing
(
2965 target_vim
, existing_instance
2968 # Specific extra params for recreate tasks:
2970 "created_items": created_items
,
2971 "vim_id": existing_instance
["vim-id"],
2972 "volumes_to_hold": volumes_to_hold
,
2975 changes_list
.append(
2977 "deployment_info": deployment_info
,
2978 "target_id": target_vim
,
2981 "target_record": target_record
,
2982 "target_record_id": target_record_id
,
2983 "extra_dict": extra_dict
,
2986 delete_task_id
= f
"{action_id}:{task_index}"
2989 # step 2 vdu to be created
2995 "vnfr": existing_vnf
,
2996 "vdu2cloud_init": vdu2cloud_init
,
2997 "tasks_by_target_record_id": tasks_by_target_record_id
,
2998 "logger": self
.logger
,
3001 "ro_nsr_public_key": ro_nsr_public_key
,
3005 extra_dict
= self
._process
_recreate
_vdu
_params
(
3014 # The CREATE task depens on the DELETE task
3015 extra_dict
["depends_on"] = [delete_task_id
]
3017 # Add volumes created from created_items if any
3018 # Ports should be deleted with delete task and automatically created with create task
3020 for k
, v
in created_items
.items():
3022 k_item
, _
, k_id
= k
.partition(":")
3023 if k_item
== "volume":
3025 except Exception as e
:
3027 "Error evaluating created item {}: {}".format(k
, e
)
3029 extra_dict
["previous_created_volumes"] = volumes
3032 "action_id": action_id
,
3034 "task_index": task_index
,
3036 self
._assign
_vim
(target_vim
)
3039 "deployment_info": deployment_info
,
3040 "target_id": target_vim
,
3043 "target_record": target_record
,
3044 "target_record_id": target_record_id
,
3045 "extra_dict": extra_dict
,
3047 changes_list
.append(new_item
)
3048 tasks_by_target_record_id
[target_record_id
] = new_item
3053 def recreate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3054 self
.logger
.debug("ns.recreate nsr_id={} indata={}".format(nsr_id
, indata
))
3055 # TODO: validate_input(indata, recreate_schema)
3056 action_id
= indata
.get("action_id", str(uuid4()))
3057 # get current deployment
3058 db_vnfrs
= {} # vnf's info indexed by _id
3060 logging_text
= "Recreate nsr_id={} action_id={} indata={}".format(
3061 nsr_id
, action_id
, indata
3063 self
.logger
.debug(logging_text
+ "Enter")
3066 step
= "Getting ns and vnfr record from db"
3067 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3069 tasks_by_target_record_id
= {}
3070 # read from db: vnf's of this ns
3071 step
= "Getting vnfrs from db"
3072 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
3073 self
.logger
.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list
))
3075 if not db_vnfrs_list
:
3076 raise NsException("Cannot obtain associated VNF for ns")
3078 for vnfr
in db_vnfrs_list
:
3079 db_vnfrs
[vnfr
["_id"]] = vnfr
3082 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
3083 self
.logger
.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr
))
3086 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
3088 with self
.write_lock
:
3090 step
= "process NS elements"
3091 changes_list
= self
.prepare_changes_to_recreate(
3096 db_ro_nsr
=db_ro_nsr
,
3097 action_id
=action_id
,
3098 tasks_by_target_record_id
=tasks_by_target_record_id
,
3101 self
.define_all_tasks(
3102 changes_list
=changes_list
,
3103 db_new_tasks
=db_new_tasks
,
3104 tasks_by_target_record_id
=tasks_by_target_record_id
,
3107 # Delete all ro_tasks registered for the targets vdurs (target_record)
3108 # If task of type CREATE exist then vim will try to get info form deleted VMs.
3109 # So remove all task related to target record.
3110 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
3111 for change
in changes_list
:
3112 for ro_task
in ro_tasks
:
3113 for task
in ro_task
["tasks"]:
3114 if task
["target_record"] == change
["target_record"]:
3118 "_id": ro_task
["_id"],
3119 "modified_at": ro_task
["modified_at"],
3121 fail_on_empty
=False,
3124 step
= "Updating database, Appending tasks to ro_tasks"
3125 self
.upload_recreate_tasks(
3126 db_new_tasks
=db_new_tasks
,
3131 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3135 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3139 except Exception as e
:
3140 if isinstance(e
, (DbException
, NsException
)):
3142 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3145 e
= traceback_format_exc()
3146 self
.logger
.critical(
3147 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3151 raise NsException(e
)
3153 def deploy(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3154 self
.logger
.debug("ns.deploy nsr_id={} indata={}".format(nsr_id
, indata
))
3155 validate_input(indata
, deploy_schema
)
3156 action_id
= indata
.get("action_id", str(uuid4()))
3158 # get current deployment
3159 db_nsr_update
= {} # update operation on nsrs
3160 db_vnfrs_update
= {}
3161 db_vnfrs
= {} # vnf's info indexed by _id
3163 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3164 self
.logger
.debug(logging_text
+ "Enter")
3167 step
= "Getting ns and vnfr record from db"
3168 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3169 self
.logger
.debug("ns.deploy: db_nsr={}".format(db_nsr
))
3171 tasks_by_target_record_id
= {}
3172 # read from db: vnf's of this ns
3173 step
= "Getting vnfrs from db"
3174 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
3176 if not db_vnfrs_list
:
3177 raise NsException("Cannot obtain associated VNF for ns")
3179 for vnfr
in db_vnfrs_list
:
3180 db_vnfrs
[vnfr
["_id"]] = vnfr
3181 db_vnfrs_update
[vnfr
["_id"]] = {}
3182 self
.logger
.debug("ns.deploy db_vnfrs={}".format(db_vnfrs
))
3185 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
3188 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
3190 # check that action_id is not in the list of actions. Suffixed with :index
3191 if action_id
in db_ro_nsr
["actions"]:
3195 new_action_id
= "{}:{}".format(action_id
, index
)
3197 if new_action_id
not in db_ro_nsr
["actions"]:
3198 action_id
= new_action_id
3201 + "Changing action_id in use to {}".format(action_id
)
3207 def _process_action(indata
):
3208 nonlocal db_new_tasks
3215 if indata
["action"]["action"] == "inject_ssh_key":
3216 key
= indata
["action"].get("key")
3217 user
= indata
["action"].get("user")
3218 password
= indata
["action"].get("password")
3220 for vnf
in indata
.get("vnf", ()):
3221 if vnf
["_id"] not in db_vnfrs
:
3222 raise NsException("Invalid vnf={}".format(vnf
["_id"]))
3224 db_vnfr
= db_vnfrs
[vnf
["_id"]]
3226 for target_vdu
in vnf
.get("vdur", ()):
3227 vdu_index
, vdur
= next(
3230 for i_v
in enumerate(db_vnfr
["vdur"])
3231 if i_v
[1]["id"] == target_vdu
["id"]
3238 "Invalid vdu vnf={}.{}".format(
3239 vnf
["_id"], target_vdu
["id"]
3243 target_vim
, vim_info
= next(
3244 k_v
for k_v
in vdur
["vim_info"].items()
3246 self
._assign
_vim
(target_vim
)
3247 target_record
= "vnfrs:{}:vdur.{}.ssh_keys".format(
3248 vnf
["_id"], vdu_index
3252 "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdur
["id"])
3255 "ip_address": vdur
.get("ip-address"),
3258 "password": password
,
3259 "private_key": db_ro_nsr
["private_key"],
3260 "salt": db_ro_nsr
["_id"],
3261 "schema_version": db_ro_nsr
["_admin"][
3268 "action_id": action_id
,
3270 "task_index": task_index
,
3273 task
= Ns
._create
_task
(
3274 deployment_info
=deployment_info
,
3275 target_id
=target_vim
,
3278 target_record
=target_record
,
3279 target_record_id
=None,
3280 extra_dict
=extra_dict
,
3283 task_index
= deployment_info
.get("task_index")
3285 db_new_tasks
.append(task
)
3287 with self
.write_lock
:
3288 if indata
.get("action"):
3289 _process_action(indata
)
3291 # compute network differences
3293 step
= "process NS elements"
3294 changes_list
= self
.calculate_all_differences_to_deploy(
3299 db_ro_nsr
=db_ro_nsr
,
3300 db_nsr_update
=db_nsr_update
,
3301 db_vnfrs_update
=db_vnfrs_update
,
3302 action_id
=action_id
,
3303 tasks_by_target_record_id
=tasks_by_target_record_id
,
3305 self
.define_all_tasks(
3306 changes_list
=changes_list
,
3307 db_new_tasks
=db_new_tasks
,
3308 tasks_by_target_record_id
=tasks_by_target_record_id
,
3311 step
= "Updating database, Appending tasks to ro_tasks"
3312 self
.upload_all_tasks(
3313 db_new_tasks
=db_new_tasks
,
3317 step
= "Updating database, nsrs"
3319 self
.db
.set_one("nsrs", {"_id": nsr_id
}, db_nsr_update
)
3321 for vnfr_id
, db_vnfr_update
in db_vnfrs_update
.items():
3323 step
= "Updating database, vnfrs={}".format(vnfr_id
)
3324 self
.db
.set_one("vnfrs", {"_id": vnfr_id
}, db_vnfr_update
)
3327 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3331 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3335 except Exception as e
:
3336 if isinstance(e
, (DbException
, NsException
)):
3338 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3341 e
= traceback_format_exc()
3342 self
.logger
.critical(
3343 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3347 raise NsException(e
)
3349 def delete(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3350 self
.logger
.debug("ns.delete version={} nsr_id={}".format(version
, nsr_id
))
3351 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
3353 with self
.write_lock
:
3355 NsWorker
.delete_db_tasks(self
.db
, nsr_id
, None)
3356 except NsWorkerException
as e
:
3357 raise NsException(e
)
3359 return None, None, True
3361 def status(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
3363 "ns.status version={} nsr_id={}, action_id={} indata={}".format(
3364 version
, nsr_id
, action_id
, indata
3370 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.action_id": action_id
})
3371 global_status
= "DONE"
3374 for ro_task
in ro_tasks
:
3375 for task
in ro_task
["tasks"]:
3376 if task
and task
["action_id"] == action_id
:
3377 task_list
.append(task
)
3380 if task
["status"] == "FAILED":
3381 global_status
= "FAILED"
3382 error_text
= "Error at {} {}: {}".format(
3383 task
["action"].lower(),
3385 ro_task
["vim_info"].get("vim_message") or "unknown",
3387 details
.append(error_text
)
3388 elif task
["status"] in ("SCHEDULED", "BUILD"):
3389 if global_status
!= "FAILED":
3390 global_status
= "BUILD"
3395 "status": global_status
,
3396 "details": ". ".join(details
)
3398 else "progress {}/{}".format(done
, total
),
3400 "action_id": action_id
,
3404 return return_data
, None, True
3406 def recreate_status(
3407 self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
3409 return self
.status(session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
)
3411 def cancel(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
3413 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
3414 session
, indata
, version
, nsr_id
, action_id
3418 return None, None, True
3420 def rebuild_start_stop_task(
3431 self
._assign
_vim
(target_vim
)
3432 target_record
= "vnfrs:{}:vdur.{}.vim_info.{}".format(
3433 vnf_id
, vdu_index
, target_vim
3435 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_id
)
3437 "action_id": action_id
,
3439 "task_index": task_index
,
3442 task
= Ns
._create
_task
(
3443 deployment_info
=deployment_info
,
3444 target_id
=target_vim
,
3447 target_record
=target_record
,
3448 target_record_id
=target_record_id
,
3449 extra_dict
=extra_dict
,
3453 def rebuild_start_stop(
3454 self
, session
, action_dict
, version
, nsr_id
, *args
, **kwargs
3459 action_id
= action_dict
.get("action_id", str(uuid4()))
3461 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3462 self
.logger
.debug(logging_text
+ "Enter")
3464 action
= list(action_dict
.keys())[0]
3465 task_dict
= action_dict
.get(action
)
3466 vim_vm_id
= action_dict
.get(action
).get("vim_vm_id")
3468 if action_dict
.get("stop"):
3472 step
= "lock the operation & do task creation"
3473 with self
.write_lock
:
3474 extra_dict
["params"] = {
3475 "vim_vm_id": vim_vm_id
,
3478 task
= self
.rebuild_start_stop_task(
3479 task_dict
["vdu_id"],
3480 task_dict
["vnf_id"],
3481 task_dict
["vdu_index"],
3485 task_dict
["target_vim"],
3488 db_new_tasks
.append(task
)
3489 step
= "upload Task to db"
3490 self
.upload_all_tasks(
3491 db_new_tasks
=db_new_tasks
,
3495 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3498 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3502 except Exception as e
:
3503 if isinstance(e
, (DbException
, NsException
)):
3505 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3508 e
= traceback_format_exc()
3509 self
.logger
.critical(
3510 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3513 raise NsException(e
)
3515 def get_deploy(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
3516 nsrs
= self
.db
.get_list("nsrs", {})
3520 return_data
.append({"_id": ns
["_id"], "name": ns
["name"]})
3522 return return_data
, None, True
3524 def get_actions(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
3525 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
3528 for ro_task
in ro_tasks
:
3529 for task
in ro_task
["tasks"]:
3530 if task
["action_id"] not in return_data
:
3531 return_data
.append(task
["action_id"])
3533 return return_data
, None, True
3536 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3538 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
3539 self
._assign
_vim
(target_vim
)
3540 target_record
= "vnfrs:{}:vdur.{}.vim_info.{}".format(
3541 vnf
["_id"], vdu_index
, target_vim
3543 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
3545 "action_id": action_id
,
3547 "task_index": task_index
,
3550 task
= Ns
._create
_task
(
3551 deployment_info
=deployment_info
,
3552 target_id
=target_vim
,
3555 target_record
=target_record
,
3556 target_record_id
=target_record_id
,
3557 extra_dict
=extra_dict
,
3562 def migrate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3566 action_id
= indata
.get("action_id", str(uuid4()))
3568 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3569 self
.logger
.debug(logging_text
+ "Enter")
3571 vnf_instance_id
= indata
["vnfInstanceId"]
3572 step
= "Getting vnfrs from db"
3573 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
3574 vdu
= indata
.get("vdu")
3575 migrateToHost
= indata
.get("migrateToHost")
3578 with self
.write_lock
:
3580 vdu_id
= indata
["vdu"]["vduId"]
3581 vdu_count_index
= indata
["vdu"].get("vduCountIndex", 0)
3582 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3584 vdu
["vdu-id-ref"] == vdu_id
3585 and vdu
["count-index"] == vdu_count_index
3587 extra_dict
["params"] = {
3588 "vim_vm_id": vdu
["vim-id"],
3589 "migrate_host": migrateToHost
,
3590 "vdu_vim_info": vdu
["vim_info"],
3592 step
= "Creating migration task for vdu:{}".format(vdu
)
3593 task
= self
.migrate_task(
3602 db_new_tasks
.append(task
)
3606 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3607 extra_dict
["params"] = {
3608 "vim_vm_id": vdu
["vim-id"],
3609 "migrate_host": migrateToHost
,
3610 "vdu_vim_info": vdu
["vim_info"],
3612 step
= "Creating migration task for vdu:{}".format(vdu
)
3613 task
= self
.migrate_task(
3622 db_new_tasks
.append(task
)
3625 self
.upload_all_tasks(
3626 db_new_tasks
=db_new_tasks
,
3631 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3634 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3638 except Exception as e
:
3639 if isinstance(e
, (DbException
, NsException
)):
3641 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3644 e
= traceback_format_exc()
3645 self
.logger
.critical(
3646 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3649 raise NsException(e
)
3651 def verticalscale_task(
3652 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3654 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
3655 self
._assign
_vim
(target_vim
)
3656 ns_preffix
= "nsrs:{}".format(nsr_id
)
3657 flavor_text
= ns_preffix
+ ":flavor." + vdu
["ns-flavor-id"]
3658 extra_dict
["depends_on"] = [flavor_text
]
3659 extra_dict
["params"].update({"flavor_id": "TASK-" + flavor_text
})
3660 target_record
= "vnfrs:{}:vdur.{}.vim_info.{}".format(
3661 vnf
["_id"], vdu_index
, target_vim
3663 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
3665 "action_id": action_id
,
3667 "task_index": task_index
,
3670 task
= Ns
._create
_task
(
3671 deployment_info
=deployment_info
,
3672 target_id
=target_vim
,
3673 item
="verticalscale",
3675 target_record
=target_record
,
3676 target_record_id
=target_record_id
,
3677 extra_dict
=extra_dict
,
3681 def verticalscale_flavor_task(
3682 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3684 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
3685 self
._assign
_vim
(target_vim
)
3686 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3687 target_record
= "nsrs:{}:flavor.{}.vim_info.{}".format(
3688 nsr_id
, len(db_nsr
["flavor"]) - 1, target_vim
3690 target_record_id
= "nsrs:{}:flavor.{}".format(nsr_id
, len(db_nsr
["flavor"]) - 1)
3692 "action_id": action_id
,
3694 "task_index": task_index
,
3696 task
= Ns
._create
_task
(
3697 deployment_info
=deployment_info
,
3698 target_id
=target_vim
,
3701 target_record
=target_record
,
3702 target_record_id
=target_record_id
,
3703 extra_dict
=extra_dict
,
3707 def verticalscale(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3710 flavor_extra_dict
= {}
3712 action_id
= indata
.get("action_id", str(uuid4()))
3714 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3715 self
.logger
.debug(logging_text
+ "Enter")
3717 VnfFlavorData
= indata
.get("changeVnfFlavorData")
3718 vnf_instance_id
= VnfFlavorData
["vnfInstanceId"]
3719 step
= "Getting vnfrs from db"
3720 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
3721 vduid
= VnfFlavorData
["additionalParams"]["vduid"]
3722 vduCountIndex
= VnfFlavorData
["additionalParams"]["vduCountIndex"]
3723 virtualMemory
= VnfFlavorData
["additionalParams"]["virtualMemory"]
3724 numVirtualCpu
= VnfFlavorData
["additionalParams"]["numVirtualCpu"]
3725 sizeOfStorage
= VnfFlavorData
["additionalParams"]["sizeOfStorage"]
3727 "name": vduid
+ "-flv",
3728 "ram": virtualMemory
,
3729 "vcpus": numVirtualCpu
,
3730 "disk": sizeOfStorage
,
3733 "ram": virtualMemory
,
3734 "vcpus": numVirtualCpu
,
3735 "disk": sizeOfStorage
,
3737 flavor_extra_dict
["find_params"] = {"flavor_data": flavor_data
}
3738 flavor_extra_dict
["params"] = {"flavor_data": flavor_dict
}
3740 step
= "Creating Tasks for vertical scaling"
3741 with self
.write_lock
:
3742 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3744 vdu
["vdu-id-ref"] == vduid
3745 and vdu
["count-index"] == vduCountIndex
3747 extra_dict
["params"] = {
3748 "vim_vm_id": vdu
["vim-id"],
3749 "flavor_dict": flavor_dict
,
3750 "vdu-id-ref": vdu
["vdu-id-ref"],
3751 "count-index": vdu
["count-index"],
3752 "vnf_instance_id": vnf_instance_id
,
3754 task
= self
.verticalscale_flavor_task(
3763 db_new_tasks
.append(task
)
3765 task
= self
.verticalscale_task(
3774 db_new_tasks
.append(task
)
3777 self
.upload_all_tasks(
3778 db_new_tasks
=db_new_tasks
,
3782 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3785 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3789 except Exception as e
:
3790 if isinstance(e
, (DbException
, NsException
)):
3792 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3795 e
= traceback_format_exc()
3796 self
.logger
.critical(
3797 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3800 raise NsException(e
)