1 # -*- coding: utf-8 -*-
4 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
19 from http
import HTTPStatus
21 from random
import choice
as random_choice
22 from threading
import Lock
24 from traceback
import format_exc
as traceback_format_exc
25 from uuid
import uuid4
27 from cryptography
.hazmat
.backends
import default_backend
as crypto_default_backend
28 from cryptography
.hazmat
.primitives
import serialization
as crypto_serialization
29 from cryptography
.hazmat
.primitives
.asymmetric
import rsa
37 from osm_common
import (
44 version
as common_version
,
46 from osm_common
.dbbase
import DbException
47 from osm_common
.fsbase
import FsException
48 from osm_common
.msgbase
import MsgException
49 from osm_ng_ro
.ns_thread
import deep_get
, NsWorker
, NsWorkerException
50 from osm_ng_ro
.validation
import deploy_schema
, validate_input
52 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
53 min_common_version
= "0.1.16"
56 class NsException(Exception):
57 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
58 self
.http_code
= http_code
59 super(Exception, self
).__init
__(message
)
64 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
65 will provide a random one
68 # Try getting docker id. If fails, get pid
70 with
open("/proc/self/cgroup", "r") as f
:
71 text_id_
= f
.readline()
72 _
, _
, text_id
= text_id_
.rpartition("/")
73 text_id
= text_id
.replace("\n", "")[:12]
81 return "".join(random_choice("0123456789abcdef") for _
in range(12))
85 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
88 for point
in v
.split("."):
89 filled
.append(point
.zfill(8))
100 # self.operations = None
102 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
103 # If done now it will not be linked to parent not getting its handler and level
105 self
.write_lock
= None
106 self
.vims_assigned
= {}
111 def init_db(self
, target_version
):
114 def start(self
, config
):
116 Connect to database, filesystem storage, and messaging
117 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
118 :param config: Configuration of db, storage, etc
122 self
.config
["process_id"] = get_process_id() # used for HA identity
123 self
.logger
= logging
.getLogger("ro.ns")
125 # check right version of common
126 if versiontuple(common_version
) < versiontuple(min_common_version
):
128 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
129 common_version
, min_common_version
135 if config
["database"]["driver"] == "mongo":
136 self
.db
= dbmongo
.DbMongo()
137 self
.db
.db_connect(config
["database"])
138 elif config
["database"]["driver"] == "memory":
139 self
.db
= dbmemory
.DbMemory()
140 self
.db
.db_connect(config
["database"])
143 "Invalid configuration param '{}' at '[database]':'driver'".format(
144 config
["database"]["driver"]
149 if config
["storage"]["driver"] == "local":
150 self
.fs
= fslocal
.FsLocal()
151 self
.fs
.fs_connect(config
["storage"])
152 elif config
["storage"]["driver"] == "mongo":
153 self
.fs
= fsmongo
.FsMongo()
154 self
.fs
.fs_connect(config
["storage"])
155 elif config
["storage"]["driver"] is None:
159 "Invalid configuration param '{}' at '[storage]':'driver'".format(
160 config
["storage"]["driver"]
165 if config
["message"]["driver"] == "local":
166 self
.msg
= msglocal
.MsgLocal()
167 self
.msg
.connect(config
["message"])
168 elif config
["message"]["driver"] == "kafka":
169 self
.msg
= msgkafka
.MsgKafka()
170 self
.msg
.connect(config
["message"])
173 "Invalid configuration param '{}' at '[message]':'driver'".format(
174 config
["message"]["driver"]
178 # TODO load workers to deal with exising database tasks
180 self
.write_lock
= Lock()
181 except (DbException
, FsException
, MsgException
) as e
:
182 raise NsException(str(e
), http_code
=e
.http_code
)
184 def get_assigned_vims(self
):
185 return list(self
.vims_assigned
.keys())
190 self
.db
.db_disconnect()
193 self
.fs
.fs_disconnect()
196 self
.msg
.disconnect()
198 self
.write_lock
= None
199 except (DbException
, FsException
, MsgException
) as e
:
200 raise NsException(str(e
), http_code
=e
.http_code
)
202 for worker
in self
.workers
:
203 worker
.insert_task(("terminate",))
205 def _create_worker(self
):
207 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
208 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
209 return the index of the assigned worker thread. Worker threads are storead at self.workers
211 # Look for a thread in idle status
215 for i
in range(len(self
.workers
))
216 if self
.workers
[i
] and self
.workers
[i
].idle
221 if worker_id
is not None:
222 # unset idle status to avoid race conditions
223 self
.workers
[worker_id
].idle
= False
225 worker_id
= len(self
.workers
)
227 if worker_id
< self
.config
["global"]["server.ns_threads"]:
228 # create a new worker
230 NsWorker(worker_id
, self
.config
, self
.plugins
, self
.db
)
232 self
.workers
[worker_id
].start()
234 # reached maximum number of threads, assign VIM to an existing one
235 worker_id
= self
.next_worker
236 self
.next_worker
= (self
.next_worker
+ 1) % self
.config
["global"][
242 def assign_vim(self
, target_id
):
243 with self
.write_lock
:
244 return self
._assign
_vim
(target_id
)
246 def _assign_vim(self
, target_id
):
247 if target_id
not in self
.vims_assigned
:
248 worker_id
= self
.vims_assigned
[target_id
] = self
._create
_worker
()
249 self
.workers
[worker_id
].insert_task(("load_vim", target_id
))
251 def reload_vim(self
, target_id
):
252 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
253 # this is because database VIM information is cached for threads working with SDN
254 with self
.write_lock
:
255 for worker
in self
.workers
:
256 if worker
and not worker
.idle
:
257 worker
.insert_task(("reload_vim", target_id
))
259 def unload_vim(self
, target_id
):
260 with self
.write_lock
:
261 return self
._unload
_vim
(target_id
)
263 def _unload_vim(self
, target_id
):
264 if target_id
in self
.vims_assigned
:
265 worker_id
= self
.vims_assigned
[target_id
]
266 self
.workers
[worker_id
].insert_task(("unload_vim", target_id
))
267 del self
.vims_assigned
[target_id
]
269 def check_vim(self
, target_id
):
270 with self
.write_lock
:
271 if target_id
in self
.vims_assigned
:
272 worker_id
= self
.vims_assigned
[target_id
]
274 worker_id
= self
._create
_worker
()
276 worker
= self
.workers
[worker_id
]
277 worker
.insert_task(("check_vim", target_id
))
279 def unload_unused_vims(self
):
280 with self
.write_lock
:
283 for target_id
in self
.vims_assigned
:
284 if not self
.db
.get_one(
287 "target_id": target_id
,
288 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
292 vims_to_unload
.append(target_id
)
294 for target_id
in vims_to_unload
:
295 self
._unload
_vim
(target_id
)
297 def _get_cloud_init(self
, where
):
299 Not used as cloud init content is provided in the http body. This method reads cloud init from a file
300 :param where: can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
303 vnfd_id
, _
, other
= where
.partition(":")
304 _type
, _
, name
= other
.partition(":")
305 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
308 base_folder
= vnfd
["_admin"]["storage"]
309 cloud_init_file
= "{}/{}/cloud_init/{}".format(
310 base_folder
["folder"], base_folder
["pkg-dir"], name
315 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
320 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
321 cloud_init_content
= ci_file
.read()
323 cloud_init_content
= vnfd
["vdu"][int(name
)]["cloud-init"]
325 raise NsException("Mismatch descriptor for cloud init: {}".format(where
))
327 return cloud_init_content
329 def _parse_jinja2(self
, cloud_init_content
, params
, context
):
331 env
= Environment(undefined
=StrictUndefined
)
332 template
= env
.from_string(cloud_init_content
)
334 return template
.render(params
or {})
335 except UndefinedError
as e
:
337 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
338 "inside the 'additionalParamsForVnf' block".format(e
, context
)
340 except (TemplateError
, TemplateNotFound
) as e
:
342 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
347 def _create_db_ro_nsrs(self
, nsr_id
, now
):
349 key
= rsa
.generate_private_key(
350 backend
=crypto_default_backend(), public_exponent
=65537, key_size
=2048
352 private_key
= key
.private_bytes(
353 crypto_serialization
.Encoding
.PEM
,
354 crypto_serialization
.PrivateFormat
.PKCS8
,
355 crypto_serialization
.NoEncryption(),
357 public_key
= key
.public_key().public_bytes(
358 crypto_serialization
.Encoding
.OpenSSH
,
359 crypto_serialization
.PublicFormat
.OpenSSH
,
361 private_key
= private_key
.decode("utf8")
362 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
363 i
= private_key
.find("\n")
364 private_key
= "-----BEGIN RSA PRIVATE KEY-----" + private_key
[i
:]
365 public_key
= public_key
.decode("utf8")
366 except Exception as e
:
367 raise NsException("Cannot create ssh-keys: {}".format(e
))
369 schema_version
= "1.1"
370 private_key_encrypted
= self
.db
.encrypt(
371 private_key
, schema_version
=schema_version
, salt
=nsr_id
378 "schema_version": schema_version
,
380 "public_key": public_key
,
381 "private_key": private_key_encrypted
,
384 self
.db
.create("ro_nsrs", db_content
)
388 def deploy(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
389 self
.logger
.debug("ns.deploy nsr_id={} indata={}".format(nsr_id
, indata
))
390 validate_input(indata
, deploy_schema
)
391 action_id
= indata
.get("action_id", str(uuid4()))
393 # get current deployment
394 db_nsr_update
= {} # update operation on nsrs
396 db_vnfrs
= {} # vnf's info indexed by _id
397 nb_ro_tasks
= 0 # for logging
398 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
400 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
401 self
.logger
.debug(logging_text
+ "Enter")
404 step
= "Getting ns and vnfr record from db"
405 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
407 tasks_by_target_record_id
= {}
408 # read from db: vnf's of this ns
409 step
= "Getting vnfrs from db"
410 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
412 if not db_vnfrs_list
:
413 raise NsException("Cannot obtain associated VNF for ns")
415 for vnfr
in db_vnfrs_list
:
416 db_vnfrs
[vnfr
["_id"]] = vnfr
417 db_vnfrs_update
[vnfr
["_id"]] = {}
420 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
423 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
425 ro_nsr_public_key
= db_ro_nsr
["public_key"]
427 # check that action_id is not in the list of actions. Suffixed with :index
428 if action_id
in db_ro_nsr
["actions"]:
432 new_action_id
= "{}:{}".format(action_id
, index
)
434 if new_action_id
not in db_ro_nsr
["actions"]:
435 action_id
= new_action_id
438 + "Changing action_id in use to {}".format(action_id
)
457 "target_id": target_id
, # it will be removed before pushing at database
458 "action_id": action_id
,
460 "task_id": "{}:{}".format(action_id
, task_index
),
461 "status": "SCHEDULED",
464 "target_record": target_record
,
465 "target_record_id": target_record_id
,
469 task
.update(extra_dict
) # params, find_params, depends_on
475 def _create_ro_task(target_id
, task
):
480 _id
= task
["task_id"]
485 "target_id": target_id
,
488 "created_items": None,
503 def _process_image_params(target_image
, vim_info
, target_record_id
):
506 if target_image
.get("image"):
507 find_params
["filter_dict"] = {"name": target_image
.get("image")}
509 if target_image
.get("vim_image_id"):
510 find_params
["filter_dict"] = {
511 "id": target_image
.get("vim_image_id")
514 if target_image
.get("image_checksum"):
515 find_params
["filter_dict"] = {
516 "checksum": target_image
.get("image_checksum")
519 return {"find_params": find_params
}
521 def _process_flavor_params(target_flavor
, vim_info
, target_record_id
):
522 def _get_resource_allocation_params(quota_descriptor
):
524 read the quota_descriptor from vnfd and fetch the resource allocation properties from the
526 :param quota_descriptor: cpu/mem/vif/disk-io quota descriptor
527 :return: quota params for limit, reserve, shares from the descriptor object
531 if quota_descriptor
.get("limit"):
532 quota
["limit"] = int(quota_descriptor
["limit"])
534 if quota_descriptor
.get("reserve"):
535 quota
["reserve"] = int(quota_descriptor
["reserve"])
537 if quota_descriptor
.get("shares"):
538 quota
["shares"] = int(quota_descriptor
["shares"])
545 "disk": int(target_flavor
["storage-gb"]),
546 "ram": int(target_flavor
["memory-mb"]),
547 "vcpus": int(target_flavor
["vcpu-count"]),
553 for vnf
in indata
.get("vnf", []):
554 for vdur
in vnf
.get("vdur", []):
555 if vdur
.get("ns-flavor-id") == target_flavor
["id"]:
558 for storage
in target_vdur
.get("virtual-storages", []):
560 storage
.get("type-of-storage")
561 == "etsi-nfv-descriptors:ephemeral-storage"
563 flavor_data
["ephemeral"] = int(
564 storage
.get("size-of-storage", 0)
567 storage
.get("type-of-storage")
568 == "etsi-nfv-descriptors:swap-storage"
570 flavor_data
["swap"] = int(storage
.get("size-of-storage", 0))
572 if target_flavor
.get("guest-epa"):
576 if target_flavor
["guest-epa"].get("numa-node-policy"):
577 numa_node_policy
= target_flavor
["guest-epa"].get(
581 if numa_node_policy
.get("node"):
582 numa_node
= numa_node_policy
["node"][0]
584 if numa_node
.get("num-cores"):
585 numa
["cores"] = numa_node
["num-cores"]
588 if numa_node
.get("paired-threads"):
589 if numa_node
["paired-threads"].get(
592 numa
["paired-threads"] = int(
593 numa_node
["paired-threads"][
600 numa_node
["paired-threads"].get("paired-thread-ids")
602 numa
["paired-threads-id"] = []
604 for pair
in numa_node
["paired-threads"][
607 numa
["paired-threads-id"].append(
609 str(pair
["thread-a"]),
610 str(pair
["thread-b"]),
614 if numa_node
.get("num-threads"):
615 numa
["threads"] = int(numa_node
["num-threads"])
618 if numa_node
.get("memory-mb"):
619 numa
["memory"] = max(
620 int(numa_node
["memory-mb"] / 1024), 1
623 if target_flavor
["guest-epa"].get("mempage-size"):
624 extended
["mempage-size"] = target_flavor
["guest-epa"].get(
629 target_flavor
["guest-epa"].get("cpu-pinning-policy")
633 target_flavor
["guest-epa"]["cpu-pinning-policy"]
637 target_flavor
["guest-epa"].get(
638 "cpu-thread-pinning-policy"
640 and target_flavor
["guest-epa"][
641 "cpu-thread-pinning-policy"
645 numa
["cores"] = max(flavor_data
["vcpus"], 1)
647 numa
["threads"] = max(flavor_data
["vcpus"], 1)
651 if target_flavor
["guest-epa"].get("cpu-quota") and not epa_vcpu_set
:
652 cpuquota
= _get_resource_allocation_params(
653 target_flavor
["guest-epa"].get("cpu-quota")
657 extended
["cpu-quota"] = cpuquota
659 if target_flavor
["guest-epa"].get("mem-quota"):
660 vduquota
= _get_resource_allocation_params(
661 target_flavor
["guest-epa"].get("mem-quota")
665 extended
["mem-quota"] = vduquota
667 if target_flavor
["guest-epa"].get("disk-io-quota"):
668 diskioquota
= _get_resource_allocation_params(
669 target_flavor
["guest-epa"].get("disk-io-quota")
673 extended
["disk-io-quota"] = diskioquota
675 if target_flavor
["guest-epa"].get("vif-quota"):
676 vifquota
= _get_resource_allocation_params(
677 target_flavor
["guest-epa"].get("vif-quota")
681 extended
["vif-quota"] = vifquota
684 extended
["numas"] = [numa
]
687 flavor_data
["extended"] = extended
689 extra_dict
= {"find_params": {"flavor_data": flavor_data
}}
690 flavor_data_name
= flavor_data
.copy()
691 flavor_data_name
["name"] = target_flavor
["name"]
692 extra_dict
["params"] = {"flavor_data": flavor_data_name
}
696 def _process_affinity_group_params(
697 target_affinity_group
, vim_info
, target_record_id
701 affinity_group_data
= {
702 "name": target_affinity_group
["name"],
703 "type": target_affinity_group
["type"],
704 "scope": target_affinity_group
["scope"],
707 extra_dict
["params"] = {
708 "affinity_group_data": affinity_group_data
,
713 def _ip_profile_2_ro(ip_profile
):
719 if "v4" in ip_profile
.get("ip-version", "ipv4")
721 "subnet_address": ip_profile
.get("subnet-address"),
722 "gateway_address": ip_profile
.get("gateway-address"),
723 "dhcp_enabled": ip_profile
.get("dhcp-params", {}).get(
726 "dhcp_start_address": ip_profile
.get("dhcp-params", {}).get(
727 "start-address", None
729 "dhcp_count": ip_profile
.get("dhcp-params", {}).get("count", None),
732 if ip_profile
.get("dns-server"):
733 ro_ip_profile
["dns_address"] = ";".join(
734 [v
["address"] for v
in ip_profile
["dns-server"]]
737 if ip_profile
.get("security-group"):
738 ro_ip_profile
["security_group"] = ip_profile
["security-group"]
742 def _process_net_params(target_vld
, vim_info
, target_record_id
):
746 if vim_info
.get("sdn"):
747 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
748 # ns_preffix = "nsrs:{}".format(nsr_id)
749 # remove the ending ".sdn
750 vld_target_record_id
, _
, _
= target_record_id
.rpartition(".")
751 extra_dict
["params"] = {
753 for k
in ("sdn-ports", "target_vim", "vlds", "type")
757 # TODO needed to add target_id in the dependency.
758 if vim_info
.get("target_vim"):
759 extra_dict
["depends_on"] = [
760 vim_info
.get("target_vim") + " " + vld_target_record_id
765 if vim_info
.get("vim_network_name"):
766 extra_dict
["find_params"] = {
767 "filter_dict": {"name": vim_info
.get("vim_network_name")}
769 elif vim_info
.get("vim_network_id"):
770 extra_dict
["find_params"] = {
771 "filter_dict": {"id": vim_info
.get("vim_network_id")}
773 elif target_vld
.get("mgmt-network"):
774 extra_dict
["find_params"] = {"mgmt": True, "name": target_vld
["id"]}
777 extra_dict
["params"] = {
778 "net_name": "{}-{}".format(
780 target_vld
.get("name", target_vld
["id"])[:16],
782 "ip_profile": _ip_profile_2_ro(vim_info
.get("ip_profile")),
783 "provider_network_profile": vim_info
.get("provider_network"),
786 if not target_vld
.get("underlay"):
787 extra_dict
["params"]["net_type"] = "bridge"
789 extra_dict
["params"]["net_type"] = (
790 "ptp" if target_vld
.get("type") == "ELINE" else "data"
795 def _process_vdu_params(target_vdu
, vim_info
, target_record_id
):
800 nonlocal vdu2cloud_init
801 nonlocal tasks_by_target_record_id
803 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
804 ns_preffix
= "nsrs:{}".format(nsr_id
)
805 image_text
= ns_preffix
+ ":image." + target_vdu
["ns-image-id"]
806 flavor_text
= ns_preffix
+ ":flavor." + target_vdu
["ns-flavor-id"]
807 extra_dict
= {"depends_on": [image_text
, flavor_text
]}
810 for iface_index
, interface
in enumerate(target_vdu
["interfaces"]):
811 if interface
.get("ns-vld-id"):
812 net_text
= ns_preffix
+ ":vld." + interface
["ns-vld-id"]
813 elif interface
.get("vnf-vld-id"):
814 net_text
= vnf_preffix
+ ":vld." + interface
["vnf-vld-id"]
817 "Interface {} from vdu {} not connected to any vld".format(
818 iface_index
, target_vdu
["vdu-name"]
822 continue # interface not connected to any vld
824 extra_dict
["depends_on"].append(net_text
)
826 if "port-security-enabled" in interface
:
827 interface
["port_security"] = interface
.pop(
828 "port-security-enabled"
831 if "port-security-disable-strategy" in interface
:
832 interface
["port_security_disable_strategy"] = interface
.pop(
833 "port-security-disable-strategy"
838 for x
, v
in interface
.items()
844 "port_security_disable_strategy",
848 net_item
["net_id"] = "TASK-" + net_text
849 net_item
["type"] = "virtual"
851 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
852 # TODO floating_ip: True/False (or it can be None)
853 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
854 # mark the net create task as type data
856 tasks_by_target_record_id
, net_text
, "params", "net_type"
858 tasks_by_target_record_id
[net_text
]["params"][
862 net_item
["use"] = "data"
863 net_item
["model"] = interface
["type"]
864 net_item
["type"] = interface
["type"]
866 interface
.get("type") == "OM-MGMT"
867 or interface
.get("mgmt-interface")
868 or interface
.get("mgmt-vnf")
870 net_item
["use"] = "mgmt"
872 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
873 net_item
["use"] = "bridge"
874 net_item
["model"] = interface
.get("type")
876 if interface
.get("ip-address"):
877 net_item
["ip_address"] = interface
["ip-address"]
879 if interface
.get("mac-address"):
880 net_item
["mac_address"] = interface
["mac-address"]
882 net_list
.append(net_item
)
884 if interface
.get("mgmt-vnf"):
885 extra_dict
["mgmt_vnf_interface"] = iface_index
886 elif interface
.get("mgmt-interface"):
887 extra_dict
["mgmt_vdu_interface"] = iface_index
892 if target_vdu
.get("cloud-init"):
893 if target_vdu
["cloud-init"] not in vdu2cloud_init
:
894 vdu2cloud_init
[target_vdu
["cloud-init"]] = self
._get
_cloud
_init
(
895 target_vdu
["cloud-init"]
898 cloud_content_
= vdu2cloud_init
[target_vdu
["cloud-init"]]
899 cloud_config
["user-data"] = self
._parse
_jinja
2(
901 target_vdu
.get("additionalParams"),
902 target_vdu
["cloud-init"],
905 if target_vdu
.get("boot-data-drive"):
906 cloud_config
["boot-data-drive"] = target_vdu
.get("boot-data-drive")
910 if target_vdu
.get("ssh-keys"):
911 ssh_keys
+= target_vdu
.get("ssh-keys")
913 if target_vdu
.get("ssh-access-required"):
914 ssh_keys
.append(ro_nsr_public_key
)
917 cloud_config
["key-pairs"] = ssh_keys
920 if target_vdu
.get("virtual-storages"):
922 {"size": disk
["size-of-storage"]}
923 for disk
in target_vdu
["virtual-storages"]
924 if disk
.get("type-of-storage")
925 == "persistent-storage:persistent-storage"
928 affinity_group_list
= []
930 if target_vdu
.get("affinity-or-anti-affinity-group-id"):
932 for affinity_group_id
in target_vdu
[
933 "affinity-or-anti-affinity-group-id"
935 affinity_group_text
= (
937 + ":affinity-or-anti-affinity-group."
941 extra_dict
["depends_on"].append(affinity_group_text
)
942 affinity_group
["affinity_group_id"] = (
943 "TASK-" + affinity_group_text
945 affinity_group_list
.append(affinity_group
)
947 extra_dict
["params"] = {
948 "name": "{}-{}-{}-{}".format(
950 vnfr
["member-vnf-index-ref"][:16],
951 target_vdu
["vdu-name"][:32],
952 target_vdu
.get("count-index") or 0,
954 "description": target_vdu
["vdu-name"],
956 "image_id": "TASK-" + image_text
,
957 "flavor_id": "TASK-" + flavor_text
,
958 "affinity_group_list": affinity_group_list
,
959 "net_list": net_list
,
960 "cloud_config": cloud_config
or None,
961 "disk_list": disk_list
,
962 "availability_zone_index": None, # TODO
963 "availability_zone_list": None, # TODO
977 nonlocal db_new_tasks
978 nonlocal tasks_by_target_record_id
981 # ensure all the target_list elements has an "id". If not assign the index as id
982 for target_index
, tl
in enumerate(target_list
):
983 if tl
and not tl
.get("id"):
984 tl
["id"] = str(target_index
)
986 # step 1 items (networks,vdus,...) to be deleted/updated
987 for item_index
, existing_item
in enumerate(existing_list
):
989 (t
for t
in target_list
if t
["id"] == existing_item
["id"]), None
992 for target_vim
, existing_viminfo
in existing_item
.get(
995 if existing_viminfo
is None:
999 target_viminfo
= target_item
.get("vim_info", {}).get(
1003 target_viminfo
= None
1005 if target_viminfo
is None:
1007 self
._assign
_vim
(target_vim
)
1008 target_record_id
= "{}.{}".format(
1009 db_record
, existing_item
["id"]
1013 if target_vim
.startswith("sdn"):
1014 # item must be sdn-net instead of net if target_vim is a sdn
1016 target_record_id
+= ".sdn"
1018 task
= _create_task(
1022 target_record
="{}.{}.vim_info.{}".format(
1023 db_record
, item_index
, target_vim
1025 target_record_id
=target_record_id
,
1027 tasks_by_target_record_id
[target_record_id
] = task
1028 db_new_tasks
.append(task
)
1030 # TODO check one by one the vims to be created/deleted
1032 # step 2 items (networks,vdus,...) to be created
1033 for target_item
in target_list
:
1036 for item_index
, existing_item
in enumerate(existing_list
):
1037 if existing_item
["id"] == target_item
["id"]:
1041 db_update
[db_path
+ ".{}".format(item_index
)] = target_item
1042 existing_list
.append(target_item
)
1043 existing_item
= None
1045 for target_vim
, target_viminfo
in target_item
.get(
1048 existing_viminfo
= None
1051 existing_viminfo
= existing_item
.get("vim_info", {}).get(
1055 # TODO check if different. Delete and create???
1056 # TODO delete if not exist
1057 if existing_viminfo
is not None:
1060 target_record_id
= "{}.{}".format(db_record
, target_item
["id"])
1063 if target_vim
.startswith("sdn"):
1064 # item must be sdn-net instead of net if target_vim is a sdn
1066 target_record_id
+= ".sdn"
1068 extra_dict
= process_params(
1069 target_item
, target_viminfo
, target_record_id
1071 self
._assign
_vim
(target_vim
)
1072 task
= _create_task(
1076 target_record
="{}.{}.vim_info.{}".format(
1077 db_record
, item_index
, target_vim
1079 target_record_id
=target_record_id
,
1080 extra_dict
=extra_dict
,
1082 tasks_by_target_record_id
[target_record_id
] = task
1083 db_new_tasks
.append(task
)
1085 if target_item
.get("common_id"):
1086 task
["common_id"] = target_item
["common_id"]
1088 db_update
[db_path
+ ".{}".format(item_index
)] = target_item
1090 def _process_action(indata
):
1091 nonlocal db_new_tasks
1096 if indata
["action"]["action"] == "inject_ssh_key":
1097 key
= indata
["action"].get("key")
1098 user
= indata
["action"].get("user")
1099 password
= indata
["action"].get("password")
1101 for vnf
in indata
.get("vnf", ()):
1102 if vnf
["_id"] not in db_vnfrs
:
1103 raise NsException("Invalid vnf={}".format(vnf
["_id"]))
1105 db_vnfr
= db_vnfrs
[vnf
["_id"]]
1107 for target_vdu
in vnf
.get("vdur", ()):
1108 vdu_index
, vdur
= next(
1111 for i_v
in enumerate(db_vnfr
["vdur"])
1112 if i_v
[1]["id"] == target_vdu
["id"]
1119 "Invalid vdu vnf={}.{}".format(
1120 vnf
["_id"], target_vdu
["id"]
1124 target_vim
, vim_info
= next(
1125 k_v
for k_v
in vdur
["vim_info"].items()
1127 self
._assign
_vim
(target_vim
)
1128 target_record
= "vnfrs:{}:vdur.{}.ssh_keys".format(
1129 vnf
["_id"], vdu_index
1133 "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdur
["id"])
1136 "ip_address": vdur
.get("ip-address"),
1139 "password": password
,
1140 "private_key": db_ro_nsr
["private_key"],
1141 "salt": db_ro_nsr
["_id"],
1142 "schema_version": db_ro_nsr
["_admin"][
1147 task
= _create_task(
1151 target_record
=target_record
,
1152 target_record_id
=None,
1153 extra_dict
=extra_dict
,
1155 db_new_tasks
.append(task
)
1157 with self
.write_lock
:
1158 if indata
.get("action"):
1159 _process_action(indata
)
1161 # compute network differences
1163 step
= "process NS VLDs"
1165 target_list
=indata
["ns"]["vld"] or [],
1166 existing_list
=db_nsr
.get("vld") or [],
1167 db_record
="nsrs:{}:vld".format(nsr_id
),
1168 db_update
=db_nsr_update
,
1171 process_params
=_process_net_params
,
1174 step
= "process NS images"
1176 target_list
=indata
.get("image") or [],
1177 existing_list
=db_nsr
.get("image") or [],
1178 db_record
="nsrs:{}:image".format(nsr_id
),
1179 db_update
=db_nsr_update
,
1182 process_params
=_process_image_params
,
1185 step
= "process NS flavors"
1187 target_list
=indata
.get("flavor") or [],
1188 existing_list
=db_nsr
.get("flavor") or [],
1189 db_record
="nsrs:{}:flavor".format(nsr_id
),
1190 db_update
=db_nsr_update
,
1193 process_params
=_process_flavor_params
,
1196 step
= "process NS Affinity Groups"
1198 target_list
=indata
.get("affinity-or-anti-affinity-group") or [],
1199 existing_list
=db_nsr
.get("affinity-or-anti-affinity-group")
1201 db_record
="nsrs:{}:affinity-or-anti-affinity-group".format(
1204 db_update
=db_nsr_update
,
1205 db_path
="affinity-or-anti-affinity-group",
1206 item
="affinity-or-anti-affinity-group",
1207 process_params
=_process_affinity_group_params
,
1211 for vnfr_id
, vnfr
in db_vnfrs
.items():
1212 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
1213 step
= "process VNF={} VLDs".format(vnfr_id
)
1217 for vnf
in indata
.get("vnf", ())
1218 if vnf
["_id"] == vnfr_id
1222 target_list
= target_vnf
.get("vld") if target_vnf
else None
1224 target_list
=target_list
or [],
1225 existing_list
=vnfr
.get("vld") or [],
1226 db_record
="vnfrs:{}:vld".format(vnfr_id
),
1227 db_update
=db_vnfrs_update
[vnfr
["_id"]],
1230 process_params
=_process_net_params
,
1233 target_list
= target_vnf
.get("vdur") if target_vnf
else None
1234 step
= "process VNF={} VDUs".format(vnfr_id
)
1236 target_list
=target_list
or [],
1237 existing_list
=vnfr
.get("vdur") or [],
1238 db_record
="vnfrs:{}:vdur".format(vnfr_id
),
1239 db_update
=db_vnfrs_update
[vnfr
["_id"]],
1242 process_params
=_process_vdu_params
,
1245 for db_task
in db_new_tasks
:
1246 step
= "Updating database, Appending tasks to ro_tasks"
1247 target_id
= db_task
.pop("target_id")
1248 common_id
= db_task
.get("common_id")
1254 "target_id": target_id
,
1255 "tasks.common_id": common_id
,
1257 update_dict
={"to_check_at": now
, "modified_at": now
},
1258 push
={"tasks": db_task
},
1259 fail_on_empty
=False,
1263 if not self
.db
.set_one(
1266 "target_id": target_id
,
1267 "tasks.target_record": db_task
["target_record"],
1269 update_dict
={"to_check_at": now
, "modified_at": now
},
1270 push
={"tasks": db_task
},
1271 fail_on_empty
=False,
1274 step
= "Updating database, Creating ro_tasks"
1275 db_ro_task
= _create_ro_task(target_id
, db_task
)
1277 self
.db
.create("ro_tasks", db_ro_task
)
1279 step
= "Updating database, nsrs"
1281 self
.db
.set_one("nsrs", {"_id": nsr_id
}, db_nsr_update
)
1283 for vnfr_id
, db_vnfr_update
in db_vnfrs_update
.items():
1285 step
= "Updating database, vnfrs={}".format(vnfr_id
)
1286 self
.db
.set_one("vnfrs", {"_id": vnfr_id
}, db_vnfr_update
)
1290 + "Exit. Created {} ro_tasks; {} tasks".format(
1291 nb_ro_tasks
, len(db_new_tasks
)
1296 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
1300 except Exception as e
:
1301 if isinstance(e
, (DbException
, NsException
)):
1303 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
1306 e
= traceback_format_exc()
1307 self
.logger
.critical(
1308 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
1312 raise NsException(e
)
1314 def delete(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
1315 self
.logger
.debug("ns.delete version={} nsr_id={}".format(version
, nsr_id
))
1316 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
1318 with self
.write_lock
:
1320 NsWorker
.delete_db_tasks(self
.db
, nsr_id
, None)
1321 except NsWorkerException
as e
:
1322 raise NsException(e
)
1324 return None, None, True
1326 def status(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
1327 # self.logger.debug("ns.status version={} nsr_id={}, action_id={} indata={}"
1328 # .format(version, nsr_id, action_id, indata))
1332 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.action_id": action_id
})
1333 global_status
= "DONE"
1336 for ro_task
in ro_tasks
:
1337 for task
in ro_task
["tasks"]:
1338 if task
and task
["action_id"] == action_id
:
1339 task_list
.append(task
)
1342 if task
["status"] == "FAILED":
1343 global_status
= "FAILED"
1344 error_text
= "Error at {} {}: {}".format(
1345 task
["action"].lower(),
1347 ro_task
["vim_info"].get("vim_details") or "unknown",
1349 details
.append(error_text
)
1350 elif task
["status"] in ("SCHEDULED", "BUILD"):
1351 if global_status
!= "FAILED":
1352 global_status
= "BUILD"
1357 "status": global_status
,
1358 "details": ". ".join(details
)
1360 else "progress {}/{}".format(done
, total
),
1362 "action_id": action_id
,
1366 return return_data
, None, True
1368 def cancel(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
1370 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
1371 session
, indata
, version
, nsr_id
, action_id
1375 return None, None, True
1377 def get_deploy(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
1378 nsrs
= self
.db
.get_list("nsrs", {})
1382 return_data
.append({"_id": ns
["_id"], "name": ns
["name"]})
1384 return return_data
, None, True
1386 def get_actions(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
1387 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
1390 for ro_task
in ro_tasks
:
1391 for task
in ro_task
["tasks"]:
1392 if task
["action_id"] not in return_data
:
1393 return_data
.append(task
["action_id"])
1395 return return_data
, None, True