1 # -*- coding: utf-8 -*-
4 # Copyright 2018 Telefonica S.A.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
20 from typing
import Any
, Dict
, List
23 import logging
.handlers
34 from osm_lcm
import ROclient
35 from osm_lcm
.data_utils
.nsr
import (
38 get_deployed_vca_list
,
41 from osm_lcm
.data_utils
.vca
import (
50 from osm_lcm
.ng_ro
import NgRoClient
, NgRoException
51 from osm_lcm
.lcm_utils
import (
59 from osm_lcm
.data_utils
.nsd
import (
60 get_ns_configuration_relation_list
,
64 from osm_lcm
.data_utils
.vnfd
import (
68 get_ee_sorted_initial_config_primitive_list
,
69 get_ee_sorted_terminate_config_primitive_list
,
71 get_virtual_link_profiles
,
76 get_number_of_instances
,
78 get_kdu_resource_profile
,
80 from osm_lcm
.data_utils
.list_utils
import find_in_list
81 from osm_lcm
.data_utils
.vnfr
import get_osm_params
, get_vdur_index
, get_kdur
82 from osm_lcm
.data_utils
.dict_utils
import parse_yaml_strings
83 from osm_lcm
.data_utils
.database
.vim_account
import VimAccountDB
84 from n2vc
.definitions
import RelationEndpoint
85 from n2vc
.k8s_helm_conn
import K8sHelmConnector
86 from n2vc
.k8s_helm3_conn
import K8sHelm3Connector
87 from n2vc
.k8s_juju_conn
import K8sJujuConnector
89 from osm_common
.dbbase
import DbException
90 from osm_common
.fsbase
import FsException
92 from osm_lcm
.data_utils
.database
.database
import Database
93 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
95 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
96 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
98 from osm_lcm
.lcm_helm_conn
import LCMHelmConn
99 from osm_lcm
.prometheus
import parse_job
101 from copy
import copy
, deepcopy
102 from time
import time
103 from uuid
import uuid4
105 from random
import randint
107 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
110 class NsLcm(LcmBase
):
111 timeout_vca_on_error
= (
113 ) # Time for charm from first time at blocked,error status to mark as failed
114 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
115 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
116 timeout_charm_delete
= 10 * 60
117 timeout_primitive
= 30 * 60 # timeout for primitive execution
118 timeout_progress_primitive
= (
120 ) # timeout for some progress in a primitive execution
122 SUBOPERATION_STATUS_NOT_FOUND
= -1
123 SUBOPERATION_STATUS_NEW
= -2
124 SUBOPERATION_STATUS_SKIP
= -3
125 task_name_deploy_vca
= "Deploying VCA"
127 def __init__(self
, msg
, lcm_tasks
, config
, loop
):
129 Init, Connect to database, filesystem storage, and messaging
130 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
133 super().__init
__(msg
=msg
, logger
=logging
.getLogger("lcm.ns"))
135 self
.db
= Database().instance
.db
136 self
.fs
= Filesystem().instance
.fs
138 self
.lcm_tasks
= lcm_tasks
139 self
.timeout
= config
["timeout"]
140 self
.ro_config
= config
["ro_config"]
141 self
.ng_ro
= config
["ro_config"].get("ng")
142 self
.vca_config
= config
["VCA"].copy()
144 # create N2VC connector
145 self
.n2vc
= N2VCJujuConnector(
148 on_update_db
=self
._on
_update
_n
2vc
_db
,
153 self
.conn_helm_ee
= LCMHelmConn(
156 vca_config
=self
.vca_config
,
157 on_update_db
=self
._on
_update
_n
2vc
_db
,
160 self
.k8sclusterhelm2
= K8sHelmConnector(
161 kubectl_command
=self
.vca_config
.get("kubectlpath"),
162 helm_command
=self
.vca_config
.get("helmpath"),
169 self
.k8sclusterhelm3
= K8sHelm3Connector(
170 kubectl_command
=self
.vca_config
.get("kubectlpath"),
171 helm_command
=self
.vca_config
.get("helm3path"),
178 self
.k8sclusterjuju
= K8sJujuConnector(
179 kubectl_command
=self
.vca_config
.get("kubectlpath"),
180 juju_command
=self
.vca_config
.get("jujupath"),
183 on_update_db
=self
._on
_update
_k
8s
_db
,
188 self
.k8scluster_map
= {
189 "helm-chart": self
.k8sclusterhelm2
,
190 "helm-chart-v3": self
.k8sclusterhelm3
,
191 "chart": self
.k8sclusterhelm3
,
192 "juju-bundle": self
.k8sclusterjuju
,
193 "juju": self
.k8sclusterjuju
,
197 "lxc_proxy_charm": self
.n2vc
,
198 "native_charm": self
.n2vc
,
199 "k8s_proxy_charm": self
.n2vc
,
200 "helm": self
.conn_helm_ee
,
201 "helm-v3": self
.conn_helm_ee
,
205 self
.RO
= NgRoClient(self
.loop
, **self
.ro_config
)
208 def increment_ip_mac(ip_mac
, vm_index
=1):
209 if not isinstance(ip_mac
, str):
212 # try with ipv4 look for last dot
213 i
= ip_mac
.rfind(".")
216 return "{}{}".format(ip_mac
[:i
], int(ip_mac
[i
:]) + vm_index
)
217 # try with ipv6 or mac look for last colon. Operate in hex
218 i
= ip_mac
.rfind(":")
221 # format in hex, len can be 2 for mac or 4 for ipv6
222 return ("{}{:0" + str(len(ip_mac
) - i
) + "x}").format(
223 ip_mac
[:i
], int(ip_mac
[i
:], 16) + vm_index
229 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
231 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
234 # TODO filter RO descriptor fields...
238 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
239 db_dict
["deploymentStatus"] = ro_descriptor
240 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
242 except Exception as e
:
244 "Cannot write database RO deployment for ns={} -> {}".format(nsrs_id
, e
)
247 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
, vca_id
=None):
249 # remove last dot from path (if exists)
250 if path
.endswith("."):
253 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
254 # .format(table, filter, path, updated_data))
257 nsr_id
= filter.get("_id")
259 # read ns record from database
260 nsr
= self
.db
.get_one(table
="nsrs", q_filter
=filter)
261 current_ns_status
= nsr
.get("nsState")
263 # get vca status for NS
264 status_dict
= await self
.n2vc
.get_status(
265 namespace
="." + nsr_id
, yaml_format
=False, vca_id
=vca_id
270 db_dict
["vcaStatus"] = status_dict
271 await self
.n2vc
.update_vca_status(db_dict
["vcaStatus"], vca_id
=vca_id
)
273 # update configurationStatus for this VCA
275 vca_index
= int(path
[path
.rfind(".") + 1 :])
278 target_dict
=nsr
, key_list
=("_admin", "deployed", "VCA")
280 vca_status
= vca_list
[vca_index
].get("status")
282 configuration_status_list
= nsr
.get("configurationStatus")
283 config_status
= configuration_status_list
[vca_index
].get("status")
285 if config_status
== "BROKEN" and vca_status
!= "failed":
286 db_dict
["configurationStatus"][vca_index
] = "READY"
287 elif config_status
!= "BROKEN" and vca_status
== "failed":
288 db_dict
["configurationStatus"][vca_index
] = "BROKEN"
289 except Exception as e
:
290 # not update configurationStatus
291 self
.logger
.debug("Error updating vca_index (ignore): {}".format(e
))
293 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
294 # if nsState = 'DEGRADED' check if all is OK
296 if current_ns_status
in ("READY", "DEGRADED"):
297 error_description
= ""
299 if status_dict
.get("machines"):
300 for machine_id
in status_dict
.get("machines"):
301 machine
= status_dict
.get("machines").get(machine_id
)
302 # check machine agent-status
303 if machine
.get("agent-status"):
304 s
= machine
.get("agent-status").get("status")
307 error_description
+= (
308 "machine {} agent-status={} ; ".format(
312 # check machine instance status
313 if machine
.get("instance-status"):
314 s
= machine
.get("instance-status").get("status")
317 error_description
+= (
318 "machine {} instance-status={} ; ".format(
323 if status_dict
.get("applications"):
324 for app_id
in status_dict
.get("applications"):
325 app
= status_dict
.get("applications").get(app_id
)
326 # check application status
327 if app
.get("status"):
328 s
= app
.get("status").get("status")
331 error_description
+= (
332 "application {} status={} ; ".format(app_id
, s
)
335 if error_description
:
336 db_dict
["errorDescription"] = error_description
337 if current_ns_status
== "READY" and is_degraded
:
338 db_dict
["nsState"] = "DEGRADED"
339 if current_ns_status
== "DEGRADED" and not is_degraded
:
340 db_dict
["nsState"] = "READY"
343 self
.update_db_2("nsrs", nsr_id
, db_dict
)
345 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
347 except Exception as e
:
348 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
350 async def _on_update_k8s_db(
351 self
, cluster_uuid
, kdu_instance
, filter=None, vca_id
=None
354 Updating vca status in NSR record
355 :param cluster_uuid: UUID of a k8s cluster
356 :param kdu_instance: The unique name of the KDU instance
357 :param filter: To get nsr_id
361 # self.logger.debug("_on_update_k8s_db(cluster_uuid={}, kdu_instance={}, filter={}"
362 # .format(cluster_uuid, kdu_instance, filter))
365 nsr_id
= filter.get("_id")
367 # get vca status for NS
368 vca_status
= await self
.k8sclusterjuju
.status_kdu(
371 complete_status
=True,
377 db_dict
["vcaStatus"] = {nsr_id
: vca_status
}
379 await self
.k8sclusterjuju
.update_vca_status(
380 db_dict
["vcaStatus"],
386 self
.update_db_2("nsrs", nsr_id
, db_dict
)
388 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
390 except Exception as e
:
391 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
394 def _parse_cloud_init(cloud_init_text
, additional_params
, vnfd_id
, vdu_id
):
396 env
= Environment(undefined
=StrictUndefined
)
397 template
= env
.from_string(cloud_init_text
)
398 return template
.render(additional_params
or {})
399 except UndefinedError
as e
:
401 "Variable {} at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
402 "file, must be provided in the instantiation parameters inside the "
403 "'additionalParamsForVnf/Vdu' block".format(e
, vnfd_id
, vdu_id
)
405 except (TemplateError
, TemplateNotFound
) as e
:
407 "Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".format(
412 def _get_vdu_cloud_init_content(self
, vdu
, vnfd
):
413 cloud_init_content
= cloud_init_file
= None
415 if vdu
.get("cloud-init-file"):
416 base_folder
= vnfd
["_admin"]["storage"]
417 if base_folder
["pkg-dir"]:
418 cloud_init_file
= "{}/{}/cloud_init/{}".format(
419 base_folder
["folder"],
420 base_folder
["pkg-dir"],
421 vdu
["cloud-init-file"],
424 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
425 base_folder
["folder"],
426 vdu
["cloud-init-file"],
428 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
429 cloud_init_content
= ci_file
.read()
430 elif vdu
.get("cloud-init"):
431 cloud_init_content
= vdu
["cloud-init"]
433 return cloud_init_content
434 except FsException
as e
:
436 "Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".format(
437 vnfd
["id"], vdu
["id"], cloud_init_file
, e
441 def _get_vdu_additional_params(self
, db_vnfr
, vdu_id
):
443 vdur
for vdur
in db_vnfr
.get("vdur") if vdu_id
== vdur
["vdu-id-ref"]
445 additional_params
= vdur
.get("additionalParams")
446 return parse_yaml_strings(additional_params
)
448 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
450 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
451 :param vnfd: input vnfd
452 :param new_id: overrides vnf id if provided
453 :param additionalParams: Instantiation params for VNFs provided
454 :param nsrId: Id of the NSR
455 :return: copy of vnfd
457 vnfd_RO
= deepcopy(vnfd
)
458 # remove unused by RO configuration, monitoring, scaling and internal keys
459 vnfd_RO
.pop("_id", None)
460 vnfd_RO
.pop("_admin", None)
461 vnfd_RO
.pop("monitoring-param", None)
462 vnfd_RO
.pop("scaling-group-descriptor", None)
463 vnfd_RO
.pop("kdu", None)
464 vnfd_RO
.pop("k8s-cluster", None)
466 vnfd_RO
["id"] = new_id
468 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
469 for vdu
in get_iterable(vnfd_RO
, "vdu"):
470 vdu
.pop("cloud-init-file", None)
471 vdu
.pop("cloud-init", None)
475 def ip_profile_2_RO(ip_profile
):
476 RO_ip_profile
= deepcopy(ip_profile
)
477 if "dns-server" in RO_ip_profile
:
478 if isinstance(RO_ip_profile
["dns-server"], list):
479 RO_ip_profile
["dns-address"] = []
480 for ds
in RO_ip_profile
.pop("dns-server"):
481 RO_ip_profile
["dns-address"].append(ds
["address"])
483 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
484 if RO_ip_profile
.get("ip-version") == "ipv4":
485 RO_ip_profile
["ip-version"] = "IPv4"
486 if RO_ip_profile
.get("ip-version") == "ipv6":
487 RO_ip_profile
["ip-version"] = "IPv6"
488 if "dhcp-params" in RO_ip_profile
:
489 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
492 def _get_ro_vim_id_for_vim_account(self
, vim_account
):
493 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
494 if db_vim
["_admin"]["operationalState"] != "ENABLED":
496 "VIM={} is not available. operationalState={}".format(
497 vim_account
, db_vim
["_admin"]["operationalState"]
500 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
503 def get_ro_wim_id_for_wim_account(self
, wim_account
):
504 if isinstance(wim_account
, str):
505 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
506 if db_wim
["_admin"]["operationalState"] != "ENABLED":
508 "WIM={} is not available. operationalState={}".format(
509 wim_account
, db_wim
["_admin"]["operationalState"]
512 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
517 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None, mark_delete
=False):
519 db_vdu_push_list
= []
520 db_update
= {"_admin.modified": time()}
522 for vdu_id
, vdu_count
in vdu_create
.items():
526 for vdur
in reversed(db_vnfr
["vdur"])
527 if vdur
["vdu-id-ref"] == vdu_id
533 "Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
538 for count
in range(vdu_count
):
539 vdur_copy
= deepcopy(vdur
)
540 vdur_copy
["status"] = "BUILD"
541 vdur_copy
["status-detailed"] = None
542 vdur_copy
["ip-address"] = None
543 vdur_copy
["_id"] = str(uuid4())
544 vdur_copy
["count-index"] += count
+ 1
545 vdur_copy
["id"] = "{}-{}".format(
546 vdur_copy
["vdu-id-ref"], vdur_copy
["count-index"]
548 vdur_copy
.pop("vim_info", None)
549 for iface
in vdur_copy
["interfaces"]:
550 if iface
.get("fixed-ip"):
551 iface
["ip-address"] = self
.increment_ip_mac(
552 iface
["ip-address"], count
+ 1
555 iface
.pop("ip-address", None)
556 if iface
.get("fixed-mac"):
557 iface
["mac-address"] = self
.increment_ip_mac(
558 iface
["mac-address"], count
+ 1
561 iface
.pop("mac-address", None)
564 ) # only first vdu can be managment of vnf
565 db_vdu_push_list
.append(vdur_copy
)
566 # self.logger.debug("scale out, adding vdu={}".format(vdur_copy))
568 for vdu_id
, vdu_count
in vdu_delete
.items():
570 indexes_to_delete
= [
572 for iv
in enumerate(db_vnfr
["vdur"])
573 if iv
[1]["vdu-id-ref"] == vdu_id
577 "vdur.{}.status".format(i
): "DELETING"
578 for i
in indexes_to_delete
[-vdu_count
:]
582 # it must be deleted one by one because common.db does not allow otherwise
585 for v
in reversed(db_vnfr
["vdur"])
586 if v
["vdu-id-ref"] == vdu_id
588 for vdu
in vdus_to_delete
[:vdu_count
]:
591 {"_id": db_vnfr
["_id"]},
593 pull
={"vdur": {"_id": vdu
["_id"]}},
595 db_push
= {"vdur": db_vdu_push_list
} if db_vdu_push_list
else None
596 self
.db
.set_one("vnfrs", {"_id": db_vnfr
["_id"]}, db_update
, push_list
=db_push
)
597 # modify passed dictionary db_vnfr
598 db_vnfr_
= self
.db
.get_one("vnfrs", {"_id": db_vnfr
["_id"]})
599 db_vnfr
["vdur"] = db_vnfr_
["vdur"]
601 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
603 Updates database nsr with the RO info for the created vld
604 :param ns_update_nsr: dictionary to be filled with the updated info
605 :param db_nsr: content of db_nsr. This is also modified
606 :param nsr_desc_RO: nsr descriptor from RO
607 :return: Nothing, LcmException is raised on errors
610 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
611 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
612 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
614 vld
["vim-id"] = net_RO
.get("vim_net_id")
615 vld
["name"] = net_RO
.get("vim_name")
616 vld
["status"] = net_RO
.get("status")
617 vld
["status-detailed"] = net_RO
.get("error_msg")
618 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
622 "ns_update_nsr: Not found vld={} at RO info".format(vld
["id"])
625 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
627 for db_vnfr
in db_vnfrs
.values():
628 vnfr_update
= {"status": "ERROR"}
629 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
630 if "status" not in vdur
:
631 vdur
["status"] = "ERROR"
632 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
634 vdur
["status-detailed"] = str(error_text
)
636 "vdur.{}.status-detailed".format(vdu_index
)
638 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
639 except DbException
as e
:
640 self
.logger
.error("Cannot update vnf. {}".format(e
))
642 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
644 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
645 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
646 :param nsr_desc_RO: nsr descriptor from RO
647 :return: Nothing, LcmException is raised on errors
649 for vnf_index
, db_vnfr
in db_vnfrs
.items():
650 for vnf_RO
in nsr_desc_RO
["vnfs"]:
651 if vnf_RO
["member_vnf_index"] != vnf_index
:
654 if vnf_RO
.get("ip_address"):
655 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
[
658 elif not db_vnfr
.get("ip-address"):
659 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
660 raise LcmExceptionNoMgmtIP(
661 "ns member_vnf_index '{}' has no IP address".format(
666 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
667 vdur_RO_count_index
= 0
668 if vdur
.get("pdu-type"):
670 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
671 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
673 if vdur
["count-index"] != vdur_RO_count_index
:
674 vdur_RO_count_index
+= 1
676 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
677 if vdur_RO
.get("ip_address"):
678 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
680 vdur
["ip-address"] = None
681 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
682 vdur
["name"] = vdur_RO
.get("vim_name")
683 vdur
["status"] = vdur_RO
.get("status")
684 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
685 for ifacer
in get_iterable(vdur
, "interfaces"):
686 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
687 if ifacer
["name"] == interface_RO
.get("internal_name"):
688 ifacer
["ip-address"] = interface_RO
.get(
691 ifacer
["mac-address"] = interface_RO
.get(
697 "ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
698 "from VIM info".format(
699 vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]
702 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
706 "ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
708 vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]
712 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
713 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
714 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
716 vld
["vim-id"] = net_RO
.get("vim_net_id")
717 vld
["name"] = net_RO
.get("vim_name")
718 vld
["status"] = net_RO
.get("status")
719 vld
["status-detailed"] = net_RO
.get("error_msg")
720 vnfr_update
["vld.{}".format(vld_index
)] = vld
724 "ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
729 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
734 "ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(
739 def _get_ns_config_info(self
, nsr_id
):
741 Generates a mapping between vnf,vdu elements and the N2VC id
742 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
743 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
744 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
745 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
747 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
748 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
750 ns_config_info
= {"osm-config-mapping": mapping
}
751 for vca
in vca_deployed_list
:
752 if not vca
["member-vnf-index"]:
754 if not vca
["vdu_id"]:
755 mapping
[vca
["member-vnf-index"]] = vca
["application"]
759 vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"]
761 ] = vca
["application"]
762 return ns_config_info
764 async def _instantiate_ng_ro(
781 def get_vim_account(vim_account_id
):
783 if vim_account_id
in db_vims
:
784 return db_vims
[vim_account_id
]
785 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account_id
})
786 db_vims
[vim_account_id
] = db_vim
789 # modify target_vld info with instantiation parameters
790 def parse_vld_instantiation_params(
791 target_vim
, target_vld
, vld_params
, target_sdn
793 if vld_params
.get("ip-profile"):
794 target_vld
["vim_info"][target_vim
]["ip_profile"] = vld_params
[
797 if vld_params
.get("provider-network"):
798 target_vld
["vim_info"][target_vim
]["provider_network"] = vld_params
[
801 if "sdn-ports" in vld_params
["provider-network"] and target_sdn
:
802 target_vld
["vim_info"][target_sdn
]["sdn-ports"] = vld_params
[
805 if vld_params
.get("wimAccountId"):
806 target_wim
= "wim:{}".format(vld_params
["wimAccountId"])
807 target_vld
["vim_info"][target_wim
] = {}
808 for param
in ("vim-network-name", "vim-network-id"):
809 if vld_params
.get(param
):
810 if isinstance(vld_params
[param
], dict):
811 for vim
, vim_net
in vld_params
[param
].items():
812 other_target_vim
= "vim:" + vim
814 target_vld
["vim_info"],
815 (other_target_vim
, param
.replace("-", "_")),
818 else: # isinstance str
819 target_vld
["vim_info"][target_vim
][
820 param
.replace("-", "_")
821 ] = vld_params
[param
]
822 if vld_params
.get("common_id"):
823 target_vld
["common_id"] = vld_params
.get("common_id")
825 # modify target["ns"]["vld"] with instantiation parameters to override vnf vim-account
826 def update_ns_vld_target(target
, ns_params
):
827 for vnf_params
in ns_params
.get("vnf", ()):
828 if vnf_params
.get("vimAccountId"):
832 for vnfr
in db_vnfrs
.values()
833 if vnf_params
["member-vnf-index"]
834 == vnfr
["member-vnf-index-ref"]
838 vdur
= next((vdur
for vdur
in target_vnf
.get("vdur", ())), None)
839 for a_index
, a_vld
in enumerate(target
["ns"]["vld"]):
840 target_vld
= find_in_list(
841 get_iterable(vdur
, "interfaces"),
842 lambda iface
: iface
.get("ns-vld-id") == a_vld
["name"],
845 if vnf_params
.get("vimAccountId") not in a_vld
.get(
848 target
["ns"]["vld"][a_index
].get("vim_info").update(
850 "vim:{}".format(vnf_params
["vimAccountId"]): {
851 "vim_network_name": ""
856 nslcmop_id
= db_nslcmop
["_id"]
858 "name": db_nsr
["name"],
861 "image": deepcopy(db_nsr
["image"]),
862 "flavor": deepcopy(db_nsr
["flavor"]),
863 "action_id": nslcmop_id
,
864 "cloud_init_content": {},
866 for image
in target
["image"]:
867 image
["vim_info"] = {}
868 for flavor
in target
["flavor"]:
869 flavor
["vim_info"] = {}
871 if db_nslcmop
.get("lcmOperationType") != "instantiate":
872 # get parameters of instantiation:
873 db_nslcmop_instantiate
= self
.db
.get_list(
876 "nsInstanceId": db_nslcmop
["nsInstanceId"],
877 "lcmOperationType": "instantiate",
880 ns_params
= db_nslcmop_instantiate
.get("operationParams")
882 ns_params
= db_nslcmop
.get("operationParams")
883 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
884 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
887 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
888 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
892 "mgmt-network": vld
.get("mgmt-network", False),
893 "type": vld
.get("type"),
896 "vim_network_name": vld
.get("vim-network-name"),
897 "vim_account_id": ns_params
["vimAccountId"],
901 # check if this network needs SDN assist
902 if vld
.get("pci-interfaces"):
903 db_vim
= get_vim_account(ns_params
["vimAccountId"])
904 sdnc_id
= db_vim
["config"].get("sdn-controller")
906 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
907 target_sdn
= "sdn:{}".format(sdnc_id
)
908 target_vld
["vim_info"][target_sdn
] = {
910 "target_vim": target_vim
,
912 "type": vld
.get("type"),
915 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
916 for nsd_vnf_profile
in nsd_vnf_profiles
:
917 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
918 if cp
["virtual-link-profile-id"] == vld
["id"]:
920 "member_vnf:{}.{}".format(
921 cp
["constituent-cpd-id"][0][
922 "constituent-base-element-id"
924 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
926 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
928 # check at nsd descriptor, if there is an ip-profile
930 nsd_vlp
= find_in_list(
931 get_virtual_link_profiles(nsd
),
932 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
937 and nsd_vlp
.get("virtual-link-protocol-data")
938 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
940 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
943 ip_profile_dest_data
= {}
944 if "ip-version" in ip_profile_source_data
:
945 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
948 if "cidr" in ip_profile_source_data
:
949 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
952 if "gateway-ip" in ip_profile_source_data
:
953 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
956 if "dhcp-enabled" in ip_profile_source_data
:
957 ip_profile_dest_data
["dhcp-params"] = {
958 "enabled": ip_profile_source_data
["dhcp-enabled"]
960 vld_params
["ip-profile"] = ip_profile_dest_data
962 # update vld_params with instantiation params
963 vld_instantiation_params
= find_in_list(
964 get_iterable(ns_params
, "vld"),
965 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
967 if vld_instantiation_params
:
968 vld_params
.update(vld_instantiation_params
)
969 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
970 target
["ns"]["vld"].append(target_vld
)
971 # Update the target ns_vld if vnf vim_account is overriden by instantiation params
972 update_ns_vld_target(target
, ns_params
)
974 for vnfr
in db_vnfrs
.values():
976 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
978 vnf_params
= find_in_list(
979 get_iterable(ns_params
, "vnf"),
980 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
982 target_vnf
= deepcopy(vnfr
)
983 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
984 for vld
in target_vnf
.get("vld", ()):
985 # check if connected to a ns.vld, to fill target'
986 vnf_cp
= find_in_list(
987 vnfd
.get("int-virtual-link-desc", ()),
988 lambda cpd
: cpd
.get("id") == vld
["id"],
991 ns_cp
= "member_vnf:{}.{}".format(
992 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
994 if cp2target
.get(ns_cp
):
995 vld
["target"] = cp2target
[ns_cp
]
998 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
1000 # check if this network needs SDN assist
1002 if vld
.get("pci-interfaces"):
1003 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1004 sdnc_id
= db_vim
["config"].get("sdn-controller")
1006 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
1007 target_sdn
= "sdn:{}".format(sdnc_id
)
1008 vld
["vim_info"][target_sdn
] = {
1010 "target_vim": target_vim
,
1012 "type": vld
.get("type"),
1015 # check at vnfd descriptor, if there is an ip-profile
1017 vnfd_vlp
= find_in_list(
1018 get_virtual_link_profiles(vnfd
),
1019 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
1023 and vnfd_vlp
.get("virtual-link-protocol-data")
1024 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
1026 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
1029 ip_profile_dest_data
= {}
1030 if "ip-version" in ip_profile_source_data
:
1031 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
1034 if "cidr" in ip_profile_source_data
:
1035 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
1038 if "gateway-ip" in ip_profile_source_data
:
1039 ip_profile_dest_data
[
1041 ] = ip_profile_source_data
["gateway-ip"]
1042 if "dhcp-enabled" in ip_profile_source_data
:
1043 ip_profile_dest_data
["dhcp-params"] = {
1044 "enabled": ip_profile_source_data
["dhcp-enabled"]
1047 vld_params
["ip-profile"] = ip_profile_dest_data
1048 # update vld_params with instantiation params
1050 vld_instantiation_params
= find_in_list(
1051 get_iterable(vnf_params
, "internal-vld"),
1052 lambda i_vld
: i_vld
["name"] == vld
["id"],
1054 if vld_instantiation_params
:
1055 vld_params
.update(vld_instantiation_params
)
1056 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1059 for vdur
in target_vnf
.get("vdur", ()):
1060 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1061 continue # This vdu must not be created
1062 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1064 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1067 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1068 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1071 and vdu_configuration
.get("config-access")
1072 and vdu_configuration
.get("config-access").get("ssh-access")
1074 vdur
["ssh-keys"] = ssh_keys_all
1075 vdur
["ssh-access-required"] = vdu_configuration
[
1077 ]["ssh-access"]["required"]
1080 and vnf_configuration
.get("config-access")
1081 and vnf_configuration
.get("config-access").get("ssh-access")
1082 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1084 vdur
["ssh-keys"] = ssh_keys_all
1085 vdur
["ssh-access-required"] = vnf_configuration
[
1087 ]["ssh-access"]["required"]
1088 elif ssh_keys_instantiation
and find_in_list(
1089 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1091 vdur
["ssh-keys"] = ssh_keys_instantiation
1093 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1095 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1097 if vdud
.get("cloud-init-file"):
1098 vdur
["cloud-init"] = "{}:file:{}".format(
1099 vnfd
["_id"], vdud
.get("cloud-init-file")
1101 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1102 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1103 base_folder
= vnfd
["_admin"]["storage"]
1104 if base_folder
["pkg-dir"]:
1105 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1106 base_folder
["folder"],
1107 base_folder
["pkg-dir"],
1108 vdud
.get("cloud-init-file"),
1111 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
1112 base_folder
["folder"],
1113 vdud
.get("cloud-init-file"),
1115 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
1116 target
["cloud_init_content"][
1119 elif vdud
.get("cloud-init"):
1120 vdur
["cloud-init"] = "{}:vdu:{}".format(
1121 vnfd
["_id"], get_vdu_index(vnfd
, vdur
["vdu-id-ref"])
1123 # put content at target.cloul_init_content. Avoid ng_ro read vnfd descriptor
1124 target
["cloud_init_content"][vdur
["cloud-init"]] = vdud
[
1127 vdur
["additionalParams"] = vdur
.get("additionalParams") or {}
1128 deploy_params_vdu
= self
._format
_additional
_params
(
1129 vdur
.get("additionalParams") or {}
1131 deploy_params_vdu
["OSM"] = get_osm_params(
1132 vnfr
, vdur
["vdu-id-ref"], vdur
["count-index"]
1134 vdur
["additionalParams"] = deploy_params_vdu
1137 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
1138 if target_vim
not in ns_flavor
["vim_info"]:
1139 ns_flavor
["vim_info"][target_vim
] = {}
1142 # in case alternative images are provided we must check if they should be applied
1143 # for the vim_type, modify the vim_type taking into account
1144 ns_image_id
= int(vdur
["ns-image-id"])
1145 if vdur
.get("alt-image-ids"):
1146 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1147 vim_type
= db_vim
["vim_type"]
1148 for alt_image_id
in vdur
.get("alt-image-ids"):
1149 ns_alt_image
= target
["image"][int(alt_image_id
)]
1150 if vim_type
== ns_alt_image
.get("vim-type"):
1151 # must use alternative image
1153 "use alternative image id: {}".format(alt_image_id
)
1155 ns_image_id
= alt_image_id
1156 vdur
["ns-image-id"] = ns_image_id
1158 ns_image
= target
["image"][int(ns_image_id
)]
1159 if target_vim
not in ns_image
["vim_info"]:
1160 ns_image
["vim_info"][target_vim
] = {}
1162 vdur
["vim_info"] = {target_vim
: {}}
1163 # instantiation parameters
1165 # vdu_instantiation_params = next((v for v in get_iterable(vnf_params, "vdu") if v["id"] ==
1166 # vdud["id"]), None)
1167 vdur_list
.append(vdur
)
1168 target_vnf
["vdur"] = vdur_list
1169 target
["vnf"].append(target_vnf
)
1171 desc
= await self
.RO
.deploy(nsr_id
, target
)
1172 self
.logger
.debug("RO return > {}".format(desc
))
1173 action_id
= desc
["action_id"]
1174 await self
._wait
_ng
_ro
(
1175 nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
1180 "_admin.deployed.RO.operational-status": "running",
1181 "detailed-status": " ".join(stage
),
1183 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1184 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1185 self
._write
_op
_status
(nslcmop_id
, stage
)
1187 logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
)
1191 async def _wait_ng_ro(
1200 detailed_status_old
= None
1202 start_time
= start_time
or time()
1203 while time() <= start_time
+ timeout
:
1204 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
1205 self
.logger
.debug("Wait NG RO > {}".format(desc_status
))
1206 if desc_status
["status"] == "FAILED":
1207 raise NgRoException(desc_status
["details"])
1208 elif desc_status
["status"] == "BUILD":
1210 stage
[2] = "VIM: ({})".format(desc_status
["details"])
1211 elif desc_status
["status"] == "DONE":
1213 stage
[2] = "Deployed at VIM"
1216 assert False, "ROclient.check_ns_status returns unknown {}".format(
1217 desc_status
["status"]
1219 if stage
and nslcmop_id
and stage
[2] != detailed_status_old
:
1220 detailed_status_old
= stage
[2]
1221 db_nsr_update
["detailed-status"] = " ".join(stage
)
1222 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1223 self
._write
_op
_status
(nslcmop_id
, stage
)
1224 await asyncio
.sleep(15, loop
=self
.loop
)
1225 else: # timeout_ns_deploy
1226 raise NgRoException("Timeout waiting ns to deploy")
1228 async def _terminate_ng_ro(
1229 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
1234 start_deploy
= time()
1241 "action_id": nslcmop_id
,
1243 desc
= await self
.RO
.deploy(nsr_id
, target
)
1244 action_id
= desc
["action_id"]
1245 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1246 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1249 + "ns terminate action at RO. action_id={}".format(action_id
)
1253 delete_timeout
= 20 * 60 # 20 minutes
1254 await self
._wait
_ng
_ro
(
1255 nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
1258 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1259 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1261 await self
.RO
.delete(nsr_id
)
1262 except Exception as e
:
1263 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1264 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1265 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1266 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1268 logging_text
+ "RO_action_id={} already deleted".format(action_id
)
1270 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1271 failed_detail
.append("delete conflict: {}".format(e
))
1274 + "RO_action_id={} delete conflict: {}".format(action_id
, e
)
1277 failed_detail
.append("delete error: {}".format(e
))
1280 + "RO_action_id={} delete error: {}".format(action_id
, e
)
1284 stage
[2] = "Error deleting from VIM"
1286 stage
[2] = "Deleted from VIM"
1287 db_nsr_update
["detailed-status"] = " ".join(stage
)
1288 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1289 self
._write
_op
_status
(nslcmop_id
, stage
)
1292 raise LcmException("; ".join(failed_detail
))
1295 async def instantiate_RO(
1309 :param logging_text: preffix text to use at logging
1310 :param nsr_id: nsr identity
1311 :param nsd: database content of ns descriptor
1312 :param db_nsr: database content of ns record
1313 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1315 :param db_vnfds: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1316 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1317 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1318 :return: None or exception
1321 start_deploy
= time()
1322 ns_params
= db_nslcmop
.get("operationParams")
1323 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1324 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1326 timeout_ns_deploy
= self
.timeout
.get(
1327 "ns_deploy", self
.timeout_ns_deploy
1330 # Check for and optionally request placement optimization. Database will be updated if placement activated
1331 stage
[2] = "Waiting for Placement."
1332 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1333 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1334 for vnfr
in db_vnfrs
.values():
1335 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1338 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1340 return await self
._instantiate
_ng
_ro
(
1353 except Exception as e
:
1354 stage
[2] = "ERROR deploying at VIM"
1355 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1357 "Error deploying at VIM {}".format(e
),
1358 exc_info
=not isinstance(
1361 ROclient
.ROClientException
,
1370 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1372 Wait for kdu to be up, get ip address
1373 :param logging_text: prefix use for logging
1380 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1383 while nb_tries
< 360:
1384 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1388 for x
in get_iterable(db_vnfr
, "kdur")
1389 if x
.get("kdu-name") == kdu_name
1395 "Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
)
1397 if kdur
.get("status"):
1398 if kdur
["status"] in ("READY", "ENABLED"):
1399 return kdur
.get("ip-address")
1402 "target KDU={} is in error state".format(kdu_name
)
1405 await asyncio
.sleep(10, loop
=self
.loop
)
1407 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1409 async def wait_vm_up_insert_key_ro(
1410 self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None
1413 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1414 :param logging_text: prefix use for logging
1419 :param pub_key: public ssh key to inject, None to skip
1420 :param user: user to apply the public ssh key
1424 self
.logger
.debug(logging_text
+ "Starting wait_vm_up_insert_key_ro")
1428 target_vdu_id
= None
1434 if ro_retries
>= 360: # 1 hour
1436 "Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
)
1439 await asyncio
.sleep(10, loop
=self
.loop
)
1442 if not target_vdu_id
:
1443 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1445 if not vdu_id
: # for the VNF case
1446 if db_vnfr
.get("status") == "ERROR":
1448 "Cannot inject ssh-key because target VNF is in error state"
1450 ip_address
= db_vnfr
.get("ip-address")
1456 for x
in get_iterable(db_vnfr
, "vdur")
1457 if x
.get("ip-address") == ip_address
1465 for x
in get_iterable(db_vnfr
, "vdur")
1466 if x
.get("vdu-id-ref") == vdu_id
1467 and x
.get("count-index") == vdu_index
1473 not vdur
and len(db_vnfr
.get("vdur", ())) == 1
1474 ): # If only one, this should be the target vdu
1475 vdur
= db_vnfr
["vdur"][0]
1478 "Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(
1479 vnfr_id
, vdu_id
, vdu_index
1482 # New generation RO stores information at "vim_info"
1485 if vdur
.get("vim_info"):
1487 t
for t
in vdur
["vim_info"]
1488 ) # there should be only one key
1489 ng_ro_status
= vdur
["vim_info"][target_vim
].get("vim_status")
1491 vdur
.get("pdu-type")
1492 or vdur
.get("status") == "ACTIVE"
1493 or ng_ro_status
== "ACTIVE"
1495 ip_address
= vdur
.get("ip-address")
1498 target_vdu_id
= vdur
["vdu-id-ref"]
1499 elif vdur
.get("status") == "ERROR" or ng_ro_status
== "ERROR":
1501 "Cannot inject ssh-key because target VM is in error state"
1504 if not target_vdu_id
:
1507 # inject public key into machine
1508 if pub_key
and user
:
1509 self
.logger
.debug(logging_text
+ "Inserting RO key")
1510 self
.logger
.debug("SSH > PubKey > {}".format(pub_key
))
1511 if vdur
.get("pdu-type"):
1512 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1515 ro_vm_id
= "{}-{}".format(
1516 db_vnfr
["member-vnf-index-ref"], target_vdu_id
1517 ) # TODO add vdu_index
1521 "action": "inject_ssh_key",
1525 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdur
["id"]}]}],
1527 desc
= await self
.RO
.deploy(nsr_id
, target
)
1528 action_id
= desc
["action_id"]
1529 await self
._wait
_ng
_ro
(nsr_id
, action_id
, timeout
=600)
1532 # wait until NS is deployed at RO
1534 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1535 ro_nsr_id
= deep_get(
1536 db_nsrs
, ("_admin", "deployed", "RO", "nsr_id")
1540 result_dict
= await self
.RO
.create_action(
1542 item_id_name
=ro_nsr_id
,
1544 "add_public_key": pub_key
,
1549 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1550 if not result_dict
or not isinstance(result_dict
, dict):
1552 "Unknown response from RO when injecting key"
1554 for result
in result_dict
.values():
1555 if result
.get("vim_result") == 200:
1558 raise ROclient
.ROClientException(
1559 "error injecting key: {}".format(
1560 result
.get("description")
1564 except NgRoException
as e
:
1566 "Reaching max tries injecting key. Error: {}".format(e
)
1568 except ROclient
.ROClientException
as e
:
1572 + "error injecting key: {}. Retrying until {} seconds".format(
1579 "Reaching max tries injecting key. Error: {}".format(e
)
1586 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1588 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1590 my_vca
= vca_deployed_list
[vca_index
]
1591 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1592 # vdu or kdu: no dependencies
1596 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1597 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1598 configuration_status_list
= db_nsr
["configurationStatus"]
1599 for index
, vca_deployed
in enumerate(configuration_status_list
):
1600 if index
== vca_index
:
1603 if not my_vca
.get("member-vnf-index") or (
1604 vca_deployed
.get("member-vnf-index")
1605 == my_vca
.get("member-vnf-index")
1607 internal_status
= configuration_status_list
[index
].get("status")
1608 if internal_status
== "READY":
1610 elif internal_status
== "BROKEN":
1612 "Configuration aborted because dependent charm/s has failed"
1617 # no dependencies, return
1619 await asyncio
.sleep(10)
1622 raise LcmException("Configuration aborted because dependent charm/s timeout")
1624 def get_vca_id(self
, db_vnfr
: dict, db_nsr
: dict):
1627 vca_id
= deep_get(db_vnfr
, ("vca-id",))
1629 vim_account_id
= deep_get(db_nsr
, ("instantiate_params", "vimAccountId"))
1630 vca_id
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("vca")
1633 async def instantiate_N2VC(
1650 ee_config_descriptor
,
1652 nsr_id
= db_nsr
["_id"]
1653 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1654 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1655 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1656 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1658 "collection": "nsrs",
1659 "filter": {"_id": nsr_id
},
1660 "path": db_update_entry
,
1666 element_under_configuration
= nsr_id
1670 vnfr_id
= db_vnfr
["_id"]
1671 osm_config
["osm"]["vnf_id"] = vnfr_id
1673 namespace
= "{nsi}.{ns}".format(nsi
=nsi_id
if nsi_id
else "", ns
=nsr_id
)
1675 if vca_type
== "native_charm":
1678 index_number
= vdu_index
or 0
1681 element_type
= "VNF"
1682 element_under_configuration
= vnfr_id
1683 namespace
+= ".{}-{}".format(vnfr_id
, index_number
)
1685 namespace
+= ".{}-{}".format(vdu_id
, index_number
)
1686 element_type
= "VDU"
1687 element_under_configuration
= "{}-{}".format(vdu_id
, index_number
)
1688 osm_config
["osm"]["vdu_id"] = vdu_id
1690 namespace
+= ".{}".format(kdu_name
)
1691 element_type
= "KDU"
1692 element_under_configuration
= kdu_name
1693 osm_config
["osm"]["kdu_name"] = kdu_name
1696 if base_folder
["pkg-dir"]:
1697 artifact_path
= "{}/{}/{}/{}".format(
1698 base_folder
["folder"],
1699 base_folder
["pkg-dir"],
1702 in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1707 artifact_path
= "{}/Scripts/{}/{}/".format(
1708 base_folder
["folder"],
1711 in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1716 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1718 # get initial_config_primitive_list that applies to this element
1719 initial_config_primitive_list
= config_descriptor
.get(
1720 "initial-config-primitive"
1724 "Initial config primitive list > {}".format(
1725 initial_config_primitive_list
1729 # add config if not present for NS charm
1730 ee_descriptor_id
= ee_config_descriptor
.get("id")
1731 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1732 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1733 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1737 "Initial config primitive list #2 > {}".format(
1738 initial_config_primitive_list
1741 # n2vc_redesign STEP 3.1
1742 # find old ee_id if exists
1743 ee_id
= vca_deployed
.get("ee_id")
1745 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1746 # create or register execution environment in VCA
1747 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1749 self
._write
_configuration
_status
(
1751 vca_index
=vca_index
,
1753 element_under_configuration
=element_under_configuration
,
1754 element_type
=element_type
,
1757 step
= "create execution environment"
1758 self
.logger
.debug(logging_text
+ step
)
1762 if vca_type
== "k8s_proxy_charm":
1763 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1764 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1765 namespace
=namespace
,
1766 artifact_path
=artifact_path
,
1770 elif vca_type
== "helm" or vca_type
== "helm-v3":
1771 ee_id
, credentials
= await self
.vca_map
[
1773 ].create_execution_environment(
1774 namespace
=namespace
,
1778 artifact_path
=artifact_path
,
1782 ee_id
, credentials
= await self
.vca_map
[
1784 ].create_execution_environment(
1785 namespace
=namespace
,
1791 elif vca_type
== "native_charm":
1792 step
= "Waiting to VM being up and getting IP address"
1793 self
.logger
.debug(logging_text
+ step
)
1794 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1803 credentials
= {"hostname": rw_mgmt_ip
}
1805 username
= deep_get(
1806 config_descriptor
, ("config-access", "ssh-access", "default-user")
1808 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1809 # merged. Meanwhile let's get username from initial-config-primitive
1810 if not username
and initial_config_primitive_list
:
1811 for config_primitive
in initial_config_primitive_list
:
1812 for param
in config_primitive
.get("parameter", ()):
1813 if param
["name"] == "ssh-username":
1814 username
= param
["value"]
1818 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1819 "'config-access.ssh-access.default-user'"
1821 credentials
["username"] = username
1822 # n2vc_redesign STEP 3.2
1824 self
._write
_configuration
_status
(
1826 vca_index
=vca_index
,
1827 status
="REGISTERING",
1828 element_under_configuration
=element_under_configuration
,
1829 element_type
=element_type
,
1832 step
= "register execution environment {}".format(credentials
)
1833 self
.logger
.debug(logging_text
+ step
)
1834 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1835 credentials
=credentials
,
1836 namespace
=namespace
,
1841 # for compatibility with MON/POL modules, the need model and application name at database
1842 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1843 ee_id_parts
= ee_id
.split(".")
1844 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1845 if len(ee_id_parts
) >= 2:
1846 model_name
= ee_id_parts
[0]
1847 application_name
= ee_id_parts
[1]
1848 db_nsr_update
[db_update_entry
+ "model"] = model_name
1849 db_nsr_update
[db_update_entry
+ "application"] = application_name
1851 # n2vc_redesign STEP 3.3
1852 step
= "Install configuration Software"
1854 self
._write
_configuration
_status
(
1856 vca_index
=vca_index
,
1857 status
="INSTALLING SW",
1858 element_under_configuration
=element_under_configuration
,
1859 element_type
=element_type
,
1860 other_update
=db_nsr_update
,
1863 # TODO check if already done
1864 self
.logger
.debug(logging_text
+ step
)
1866 if vca_type
== "native_charm":
1867 config_primitive
= next(
1868 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1871 if config_primitive
:
1872 config
= self
._map
_primitive
_params
(
1873 config_primitive
, {}, deploy_params
1876 if vca_type
== "lxc_proxy_charm":
1877 if element_type
== "NS":
1878 num_units
= db_nsr
.get("config-units") or 1
1879 elif element_type
== "VNF":
1880 num_units
= db_vnfr
.get("config-units") or 1
1881 elif element_type
== "VDU":
1882 for v
in db_vnfr
["vdur"]:
1883 if vdu_id
== v
["vdu-id-ref"]:
1884 num_units
= v
.get("config-units") or 1
1886 if vca_type
!= "k8s_proxy_charm":
1887 await self
.vca_map
[vca_type
].install_configuration_sw(
1889 artifact_path
=artifact_path
,
1892 num_units
=num_units
,
1897 # write in db flag of configuration_sw already installed
1899 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1902 # add relations for this VCA (wait for other peers related with this VCA)
1903 await self
._add
_vca
_relations
(
1904 logging_text
=logging_text
,
1907 vca_index
=vca_index
,
1910 # if SSH access is required, then get execution environment SSH public
1911 # if native charm we have waited already to VM be UP
1912 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1915 # self.logger.debug("get ssh key block")
1917 config_descriptor
, ("config-access", "ssh-access", "required")
1919 # self.logger.debug("ssh key needed")
1920 # Needed to inject a ssh key
1923 ("config-access", "ssh-access", "default-user"),
1925 step
= "Install configuration Software, getting public ssh key"
1926 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1927 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1930 step
= "Insert public key into VM user={} ssh_key={}".format(
1934 # self.logger.debug("no need to get ssh key")
1935 step
= "Waiting to VM being up and getting IP address"
1936 self
.logger
.debug(logging_text
+ step
)
1938 # n2vc_redesign STEP 5.1
1939 # wait for RO (ip-address) Insert pub_key into VM
1942 rw_mgmt_ip
= await self
.wait_kdu_up(
1943 logging_text
, nsr_id
, vnfr_id
, kdu_name
1946 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1956 rw_mgmt_ip
= None # This is for a NS configuration
1958 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1960 # store rw_mgmt_ip in deploy params for later replacement
1961 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1963 # n2vc_redesign STEP 6 Execute initial config primitive
1964 step
= "execute initial config primitive"
1966 # wait for dependent primitives execution (NS -> VNF -> VDU)
1967 if initial_config_primitive_list
:
1968 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1970 # stage, in function of element type: vdu, kdu, vnf or ns
1971 my_vca
= vca_deployed_list
[vca_index
]
1972 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1974 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1975 elif my_vca
.get("member-vnf-index"):
1977 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1980 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1982 self
._write
_configuration
_status
(
1983 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1986 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1988 check_if_terminated_needed
= True
1989 for initial_config_primitive
in initial_config_primitive_list
:
1990 # adding information on the vca_deployed if it is a NS execution environment
1991 if not vca_deployed
["member-vnf-index"]:
1992 deploy_params
["ns_config_info"] = json
.dumps(
1993 self
._get
_ns
_config
_info
(nsr_id
)
1995 # TODO check if already done
1996 primitive_params_
= self
._map
_primitive
_params
(
1997 initial_config_primitive
, {}, deploy_params
2000 step
= "execute primitive '{}' params '{}'".format(
2001 initial_config_primitive
["name"], primitive_params_
2003 self
.logger
.debug(logging_text
+ step
)
2004 await self
.vca_map
[vca_type
].exec_primitive(
2006 primitive_name
=initial_config_primitive
["name"],
2007 params_dict
=primitive_params_
,
2012 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
2013 if check_if_terminated_needed
:
2014 if config_descriptor
.get("terminate-config-primitive"):
2016 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
2018 check_if_terminated_needed
= False
2020 # TODO register in database that primitive is done
2022 # STEP 7 Configure metrics
2023 if vca_type
== "helm" or vca_type
== "helm-v3":
2024 prometheus_jobs
= await self
.extract_prometheus_scrape_jobs(
2026 artifact_path
=artifact_path
,
2027 ee_config_descriptor
=ee_config_descriptor
,
2030 target_ip
=rw_mgmt_ip
,
2036 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
2039 for job
in prometheus_jobs
:
2042 {"job_name": job
["job_name"]},
2045 fail_on_empty
=False,
2048 step
= "instantiated at VCA"
2049 self
.logger
.debug(logging_text
+ step
)
2051 self
._write
_configuration
_status
(
2052 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
2055 except Exception as e
: # TODO not use Exception but N2VC exception
2056 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
2058 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
2061 "Exception while {} : {}".format(step
, e
), exc_info
=True
2063 self
._write
_configuration
_status
(
2064 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
2066 raise LcmException("{} {}".format(step
, e
)) from e
2068 def _write_ns_status(
2072 current_operation
: str,
2073 current_operation_id
: str,
2074 error_description
: str = None,
2075 error_detail
: str = None,
2076 other_update
: dict = None,
2079 Update db_nsr fields.
2082 :param current_operation:
2083 :param current_operation_id:
2084 :param error_description:
2085 :param error_detail:
2086 :param other_update: Other required changes at database if provided, will be cleared
2090 db_dict
= other_update
or {}
2093 ] = current_operation_id
# for backward compatibility
2094 db_dict
["_admin.current-operation"] = current_operation_id
2095 db_dict
["_admin.operation-type"] = (
2096 current_operation
if current_operation
!= "IDLE" else None
2098 db_dict
["currentOperation"] = current_operation
2099 db_dict
["currentOperationID"] = current_operation_id
2100 db_dict
["errorDescription"] = error_description
2101 db_dict
["errorDetail"] = error_detail
2104 db_dict
["nsState"] = ns_state
2105 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2106 except DbException
as e
:
2107 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2109 def _write_op_status(
2113 error_message
: str = None,
2114 queuePosition
: int = 0,
2115 operation_state
: str = None,
2116 other_update
: dict = None,
2119 db_dict
= other_update
or {}
2120 db_dict
["queuePosition"] = queuePosition
2121 if isinstance(stage
, list):
2122 db_dict
["stage"] = stage
[0]
2123 db_dict
["detailed-status"] = " ".join(stage
)
2124 elif stage
is not None:
2125 db_dict
["stage"] = str(stage
)
2127 if error_message
is not None:
2128 db_dict
["errorMessage"] = error_message
2129 if operation_state
is not None:
2130 db_dict
["operationState"] = operation_state
2131 db_dict
["statusEnteredTime"] = time()
2132 self
.update_db_2("nslcmops", op_id
, db_dict
)
2133 except DbException
as e
:
2135 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2138 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2140 nsr_id
= db_nsr
["_id"]
2141 # configurationStatus
2142 config_status
= db_nsr
.get("configurationStatus")
2145 "configurationStatus.{}.status".format(index
): status
2146 for index
, v
in enumerate(config_status
)
2150 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2152 except DbException
as e
:
2154 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2157 def _write_configuration_status(
2162 element_under_configuration
: str = None,
2163 element_type
: str = None,
2164 other_update
: dict = None,
2167 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2168 # .format(vca_index, status))
2171 db_path
= "configurationStatus.{}.".format(vca_index
)
2172 db_dict
= other_update
or {}
2174 db_dict
[db_path
+ "status"] = status
2175 if element_under_configuration
:
2177 db_path
+ "elementUnderConfiguration"
2178 ] = element_under_configuration
2180 db_dict
[db_path
+ "elementType"] = element_type
2181 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2182 except DbException
as e
:
2184 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2185 status
, nsr_id
, vca_index
, e
2189 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2191 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2192 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2193 Database is used because the result can be obtained from a different LCM worker in case of HA.
2194 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2195 :param db_nslcmop: database content of nslcmop
2196 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2197 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2198 computed 'vim-account-id'
2201 nslcmop_id
= db_nslcmop
["_id"]
2202 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2203 if placement_engine
== "PLA":
2205 logging_text
+ "Invoke and wait for placement optimization"
2207 await self
.msg
.aiowrite(
2208 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2210 db_poll_interval
= 5
2211 wait
= db_poll_interval
* 10
2213 while not pla_result
and wait
>= 0:
2214 await asyncio
.sleep(db_poll_interval
)
2215 wait
-= db_poll_interval
2216 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2217 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2221 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2224 for pla_vnf
in pla_result
["vnf"]:
2225 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2226 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2231 {"_id": vnfr
["_id"]},
2232 {"vim-account-id": pla_vnf
["vimAccountId"]},
2235 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2238 def update_nsrs_with_pla_result(self
, params
):
2240 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2242 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2244 except Exception as e
:
2245 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2247 async def instantiate(self
, nsr_id
, nslcmop_id
):
2250 :param nsr_id: ns instance to deploy
2251 :param nslcmop_id: operation to run
2255 # Try to lock HA task here
2256 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2257 if not task_is_locked_by_me
:
2259 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2263 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2264 self
.logger
.debug(logging_text
+ "Enter")
2266 # get all needed from database
2268 # database nsrs record
2271 # database nslcmops record
2274 # update operation on nsrs
2276 # update operation on nslcmops
2277 db_nslcmop_update
= {}
2279 nslcmop_operation_state
= None
2280 db_vnfrs
= {} # vnf's info indexed by member-index
2282 tasks_dict_info
= {} # from task to info text
2286 "Stage 1/5: preparation of the environment.",
2287 "Waiting for previous operations to terminate.",
2290 # ^ stage, step, VIM progress
2292 # wait for any previous tasks in process
2293 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2295 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2296 stage
[1] = "Reading from database."
2297 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2298 db_nsr_update
["detailed-status"] = "creating"
2299 db_nsr_update
["operational-status"] = "init"
2300 self
._write
_ns
_status
(
2302 ns_state
="BUILDING",
2303 current_operation
="INSTANTIATING",
2304 current_operation_id
=nslcmop_id
,
2305 other_update
=db_nsr_update
,
2307 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2309 # read from db: operation
2310 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2311 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2312 if db_nslcmop
["operationParams"].get("additionalParamsForVnf"):
2313 db_nslcmop
["operationParams"]["additionalParamsForVnf"] = json
.loads(
2314 db_nslcmop
["operationParams"]["additionalParamsForVnf"]
2316 ns_params
= db_nslcmop
.get("operationParams")
2317 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2318 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2320 timeout_ns_deploy
= self
.timeout
.get(
2321 "ns_deploy", self
.timeout_ns_deploy
2325 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2326 self
.logger
.debug(logging_text
+ stage
[1])
2327 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2328 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2329 self
.logger
.debug(logging_text
+ stage
[1])
2330 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2331 self
.fs
.sync(db_nsr
["nsd-id"])
2333 # nsr_name = db_nsr["name"] # TODO short-name??
2335 # read from db: vnf's of this ns
2336 stage
[1] = "Getting vnfrs from db."
2337 self
.logger
.debug(logging_text
+ stage
[1])
2338 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2340 # read from db: vnfd's for every vnf
2341 db_vnfds
= [] # every vnfd data
2343 # for each vnf in ns, read vnfd
2344 for vnfr
in db_vnfrs_list
:
2345 if vnfr
.get("kdur"):
2347 for kdur
in vnfr
["kdur"]:
2348 if kdur
.get("additionalParams"):
2349 kdur
["additionalParams"] = json
.loads(kdur
["additionalParams"])
2350 kdur_list
.append(kdur
)
2351 vnfr
["kdur"] = kdur_list
2353 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2354 vnfd_id
= vnfr
["vnfd-id"]
2355 vnfd_ref
= vnfr
["vnfd-ref"]
2356 self
.fs
.sync(vnfd_id
)
2358 # if we haven't this vnfd, read it from db
2359 if vnfd_id
not in db_vnfds
:
2361 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2364 self
.logger
.debug(logging_text
+ stage
[1])
2365 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2368 db_vnfds
.append(vnfd
)
2370 # Get or generates the _admin.deployed.VCA list
2371 vca_deployed_list
= None
2372 if db_nsr
["_admin"].get("deployed"):
2373 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2374 if vca_deployed_list
is None:
2375 vca_deployed_list
= []
2376 configuration_status_list
= []
2377 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2378 db_nsr_update
["configurationStatus"] = configuration_status_list
2379 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2380 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2381 elif isinstance(vca_deployed_list
, dict):
2382 # maintain backward compatibility. Change a dict to list at database
2383 vca_deployed_list
= list(vca_deployed_list
.values())
2384 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2385 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2388 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2390 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2391 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2393 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2394 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2395 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2397 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2400 # n2vc_redesign STEP 2 Deploy Network Scenario
2401 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2402 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2404 stage
[1] = "Deploying KDUs."
2405 # self.logger.debug(logging_text + "Before deploy_kdus")
2406 # Call to deploy_kdus in case exists the "vdu:kdu" param
2407 await self
.deploy_kdus(
2408 logging_text
=logging_text
,
2410 nslcmop_id
=nslcmop_id
,
2413 task_instantiation_info
=tasks_dict_info
,
2416 stage
[1] = "Getting VCA public key."
2417 # n2vc_redesign STEP 1 Get VCA public ssh-key
2418 # feature 1429. Add n2vc public key to needed VMs
2419 n2vc_key
= self
.n2vc
.get_public_key()
2420 n2vc_key_list
= [n2vc_key
]
2421 if self
.vca_config
.get("public_key"):
2422 n2vc_key_list
.append(self
.vca_config
["public_key"])
2424 stage
[1] = "Deploying NS at VIM."
2425 task_ro
= asyncio
.ensure_future(
2426 self
.instantiate_RO(
2427 logging_text
=logging_text
,
2431 db_nslcmop
=db_nslcmop
,
2434 n2vc_key_list
=n2vc_key_list
,
2438 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2439 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2441 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2442 stage
[1] = "Deploying Execution Environments."
2443 self
.logger
.debug(logging_text
+ stage
[1])
2445 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2446 for vnf_profile
in get_vnf_profiles(nsd
):
2447 vnfd_id
= vnf_profile
["vnfd-id"]
2448 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2449 member_vnf_index
= str(vnf_profile
["id"])
2450 db_vnfr
= db_vnfrs
[member_vnf_index
]
2451 base_folder
= vnfd
["_admin"]["storage"]
2457 # Get additional parameters
2458 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2459 if db_vnfr
.get("additionalParamsForVnf"):
2460 deploy_params
.update(
2461 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2464 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2465 if descriptor_config
:
2467 logging_text
=logging_text
2468 + "member_vnf_index={} ".format(member_vnf_index
),
2471 nslcmop_id
=nslcmop_id
,
2477 member_vnf_index
=member_vnf_index
,
2478 vdu_index
=vdu_index
,
2480 deploy_params
=deploy_params
,
2481 descriptor_config
=descriptor_config
,
2482 base_folder
=base_folder
,
2483 task_instantiation_info
=tasks_dict_info
,
2487 # Deploy charms for each VDU that supports one.
2488 for vdud
in get_vdu_list(vnfd
):
2490 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2491 vdur
= find_in_list(
2492 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2495 if vdur
.get("additionalParams"):
2496 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2498 deploy_params_vdu
= deploy_params
2499 deploy_params_vdu
["OSM"] = get_osm_params(
2500 db_vnfr
, vdu_id
, vdu_count_index
=0
2502 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2504 self
.logger
.debug("VDUD > {}".format(vdud
))
2506 "Descriptor config > {}".format(descriptor_config
)
2508 if descriptor_config
:
2511 for vdu_index
in range(vdud_count
):
2512 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2514 logging_text
=logging_text
2515 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2516 member_vnf_index
, vdu_id
, vdu_index
2520 nslcmop_id
=nslcmop_id
,
2526 member_vnf_index
=member_vnf_index
,
2527 vdu_index
=vdu_index
,
2529 deploy_params
=deploy_params_vdu
,
2530 descriptor_config
=descriptor_config
,
2531 base_folder
=base_folder
,
2532 task_instantiation_info
=tasks_dict_info
,
2535 for kdud
in get_kdu_list(vnfd
):
2536 kdu_name
= kdud
["name"]
2537 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2538 if descriptor_config
:
2543 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2545 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2546 if kdur
.get("additionalParams"):
2547 deploy_params_kdu
= parse_yaml_strings(
2548 kdur
["additionalParams"]
2552 logging_text
=logging_text
,
2555 nslcmop_id
=nslcmop_id
,
2561 member_vnf_index
=member_vnf_index
,
2562 vdu_index
=vdu_index
,
2564 deploy_params
=deploy_params_kdu
,
2565 descriptor_config
=descriptor_config
,
2566 base_folder
=base_folder
,
2567 task_instantiation_info
=tasks_dict_info
,
2571 # Check if this NS has a charm configuration
2572 descriptor_config
= nsd
.get("ns-configuration")
2573 if descriptor_config
and descriptor_config
.get("juju"):
2576 member_vnf_index
= None
2582 # Get additional parameters
2583 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2584 if db_nsr
.get("additionalParamsForNs"):
2585 deploy_params
.update(
2586 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2588 base_folder
= nsd
["_admin"]["storage"]
2590 logging_text
=logging_text
,
2593 nslcmop_id
=nslcmop_id
,
2599 member_vnf_index
=member_vnf_index
,
2600 vdu_index
=vdu_index
,
2602 deploy_params
=deploy_params
,
2603 descriptor_config
=descriptor_config
,
2604 base_folder
=base_folder
,
2605 task_instantiation_info
=tasks_dict_info
,
2609 # rest of staff will be done at finally
2612 ROclient
.ROClientException
,
2618 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2621 except asyncio
.CancelledError
:
2623 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2625 exc
= "Operation was cancelled"
2626 except Exception as e
:
2627 exc
= traceback
.format_exc()
2628 self
.logger
.critical(
2629 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2634 error_list
.append(str(exc
))
2636 # wait for pending tasks
2638 stage
[1] = "Waiting for instantiate pending tasks."
2639 self
.logger
.debug(logging_text
+ stage
[1])
2640 error_list
+= await self
._wait
_for
_tasks
(
2648 stage
[1] = stage
[2] = ""
2649 except asyncio
.CancelledError
:
2650 error_list
.append("Cancelled")
2651 # TODO cancel all tasks
2652 except Exception as exc
:
2653 error_list
.append(str(exc
))
2655 # update operation-status
2656 db_nsr_update
["operational-status"] = "running"
2657 # let's begin with VCA 'configured' status (later we can change it)
2658 db_nsr_update
["config-status"] = "configured"
2659 for task
, task_name
in tasks_dict_info
.items():
2660 if not task
.done() or task
.cancelled() or task
.exception():
2661 if task_name
.startswith(self
.task_name_deploy_vca
):
2662 # A N2VC task is pending
2663 db_nsr_update
["config-status"] = "failed"
2665 # RO or KDU task is pending
2666 db_nsr_update
["operational-status"] = "failed"
2668 # update status at database
2670 error_detail
= ". ".join(error_list
)
2671 self
.logger
.error(logging_text
+ error_detail
)
2672 error_description_nslcmop
= "{} Detail: {}".format(
2673 stage
[0], error_detail
2675 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2676 nslcmop_id
, stage
[0]
2679 db_nsr_update
["detailed-status"] = (
2680 error_description_nsr
+ " Detail: " + error_detail
2682 db_nslcmop_update
["detailed-status"] = error_detail
2683 nslcmop_operation_state
= "FAILED"
2687 error_description_nsr
= error_description_nslcmop
= None
2689 db_nsr_update
["detailed-status"] = "Done"
2690 db_nslcmop_update
["detailed-status"] = "Done"
2691 nslcmop_operation_state
= "COMPLETED"
2694 self
._write
_ns
_status
(
2697 current_operation
="IDLE",
2698 current_operation_id
=None,
2699 error_description
=error_description_nsr
,
2700 error_detail
=error_detail
,
2701 other_update
=db_nsr_update
,
2703 self
._write
_op
_status
(
2706 error_message
=error_description_nslcmop
,
2707 operation_state
=nslcmop_operation_state
,
2708 other_update
=db_nslcmop_update
,
2711 if nslcmop_operation_state
:
2713 await self
.msg
.aiowrite(
2718 "nslcmop_id": nslcmop_id
,
2719 "operationState": nslcmop_operation_state
,
2723 except Exception as e
:
2725 logging_text
+ "kafka_write notification Exception {}".format(e
)
2728 self
.logger
.debug(logging_text
+ "Exit")
2729 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2731 def _get_vnfd(self
, vnfd_id
: str, cached_vnfds
: Dict
[str, Any
]):
2732 if vnfd_id
not in cached_vnfds
:
2733 cached_vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
2734 return cached_vnfds
[vnfd_id
]
2736 def _get_vnfr(self
, nsr_id
: str, vnf_profile_id
: str, cached_vnfrs
: Dict
[str, Any
]):
2737 if vnf_profile_id
not in cached_vnfrs
:
2738 cached_vnfrs
[vnf_profile_id
] = self
.db
.get_one(
2741 "member-vnf-index-ref": vnf_profile_id
,
2742 "nsr-id-ref": nsr_id
,
2745 return cached_vnfrs
[vnf_profile_id
]
2747 def _is_deployed_vca_in_relation(
2748 self
, vca
: DeployedVCA
, relation
: Relation
2751 for endpoint
in (relation
.provider
, relation
.requirer
):
2752 if endpoint
["kdu-resource-profile-id"]:
2755 vca
.vnf_profile_id
== endpoint
.vnf_profile_id
2756 and vca
.vdu_profile_id
== endpoint
.vdu_profile_id
2757 and vca
.execution_environment_ref
== endpoint
.execution_environment_ref
2763 def _update_ee_relation_data_with_implicit_data(
2764 self
, nsr_id
, nsd
, ee_relation_data
, cached_vnfds
, vnf_profile_id
: str = None
2766 ee_relation_data
= safe_get_ee_relation(
2767 nsr_id
, ee_relation_data
, vnf_profile_id
=vnf_profile_id
2769 ee_relation_level
= EELevel
.get_level(ee_relation_data
)
2770 if (ee_relation_level
in (EELevel
.VNF
, EELevel
.VDU
)) and not ee_relation_data
[
2771 "execution-environment-ref"
2773 vnf_profile
= get_vnf_profile(nsd
, ee_relation_data
["vnf-profile-id"])
2774 vnfd_id
= vnf_profile
["vnfd-id"]
2775 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2778 if ee_relation_level
== EELevel
.VNF
2779 else ee_relation_data
["vdu-profile-id"]
2781 ee
= get_juju_ee_ref(db_vnfd
, entity_id
)
2784 f
"not execution environments found for ee_relation {ee_relation_data}"
2786 ee_relation_data
["execution-environment-ref"] = ee
["id"]
2787 return ee_relation_data
2789 def _get_ns_relations(
2792 nsd
: Dict
[str, Any
],
2794 cached_vnfds
: Dict
[str, Any
],
2795 ) -> List
[Relation
]:
2797 db_ns_relations
= get_ns_configuration_relation_list(nsd
)
2798 for r
in db_ns_relations
:
2799 provider_dict
= None
2800 requirer_dict
= None
2801 if all(key
in r
for key
in ("provider", "requirer")):
2802 provider_dict
= r
["provider"]
2803 requirer_dict
= r
["requirer"]
2804 elif "entities" in r
:
2805 provider_id
= r
["entities"][0]["id"]
2808 "endpoint": r
["entities"][0]["endpoint"],
2810 if provider_id
!= nsd
["id"]:
2811 provider_dict
["vnf-profile-id"] = provider_id
2812 requirer_id
= r
["entities"][1]["id"]
2815 "endpoint": r
["entities"][1]["endpoint"],
2817 if requirer_id
!= nsd
["id"]:
2818 requirer_dict
["vnf-profile-id"] = requirer_id
2821 "provider/requirer or entities must be included in the relation."
2823 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2824 nsr_id
, nsd
, provider_dict
, cached_vnfds
2826 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2827 nsr_id
, nsd
, requirer_dict
, cached_vnfds
2829 provider
= EERelation(relation_provider
)
2830 requirer
= EERelation(relation_requirer
)
2831 relation
= Relation(r
["name"], provider
, requirer
)
2832 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2834 relations
.append(relation
)
2837 def _get_vnf_relations(
2840 nsd
: Dict
[str, Any
],
2842 cached_vnfds
: Dict
[str, Any
],
2843 ) -> List
[Relation
]:
2845 vnf_profile
= get_vnf_profile(nsd
, vca
.vnf_profile_id
)
2846 vnf_profile_id
= vnf_profile
["id"]
2847 vnfd_id
= vnf_profile
["vnfd-id"]
2848 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2849 db_vnf_relations
= get_relation_list(db_vnfd
, vnfd_id
)
2850 for r
in db_vnf_relations
:
2851 provider_dict
= None
2852 requirer_dict
= None
2853 if all(key
in r
for key
in ("provider", "requirer")):
2854 provider_dict
= r
["provider"]
2855 requirer_dict
= r
["requirer"]
2856 elif "entities" in r
:
2857 provider_id
= r
["entities"][0]["id"]
2860 "vnf-profile-id": vnf_profile_id
,
2861 "endpoint": r
["entities"][0]["endpoint"],
2863 if provider_id
!= vnfd_id
:
2864 provider_dict
["vdu-profile-id"] = provider_id
2865 requirer_id
= r
["entities"][1]["id"]
2868 "vnf-profile-id": vnf_profile_id
,
2869 "endpoint": r
["entities"][1]["endpoint"],
2871 if requirer_id
!= vnfd_id
:
2872 requirer_dict
["vdu-profile-id"] = requirer_id
2875 "provider/requirer or entities must be included in the relation."
2877 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2878 nsr_id
, nsd
, provider_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2880 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2881 nsr_id
, nsd
, requirer_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2883 provider
= EERelation(relation_provider
)
2884 requirer
= EERelation(relation_requirer
)
2885 relation
= Relation(r
["name"], provider
, requirer
)
2886 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2888 relations
.append(relation
)
2891 def _get_kdu_resource_data(
2893 ee_relation
: EERelation
,
2894 db_nsr
: Dict
[str, Any
],
2895 cached_vnfds
: Dict
[str, Any
],
2896 ) -> DeployedK8sResource
:
2897 nsd
= get_nsd(db_nsr
)
2898 vnf_profiles
= get_vnf_profiles(nsd
)
2899 vnfd_id
= find_in_list(
2901 lambda vnf_profile
: vnf_profile
["id"] == ee_relation
.vnf_profile_id
,
2903 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2904 kdu_resource_profile
= get_kdu_resource_profile(
2905 db_vnfd
, ee_relation
.kdu_resource_profile_id
2907 kdu_name
= kdu_resource_profile
["kdu-name"]
2908 deployed_kdu
, _
= get_deployed_kdu(
2909 db_nsr
.get("_admin", ()).get("deployed", ()),
2911 ee_relation
.vnf_profile_id
,
2913 deployed_kdu
.update({"resource-name": kdu_resource_profile
["resource-name"]})
2916 def _get_deployed_component(
2918 ee_relation
: EERelation
,
2919 db_nsr
: Dict
[str, Any
],
2920 cached_vnfds
: Dict
[str, Any
],
2921 ) -> DeployedComponent
:
2922 nsr_id
= db_nsr
["_id"]
2923 deployed_component
= None
2924 ee_level
= EELevel
.get_level(ee_relation
)
2925 if ee_level
== EELevel
.NS
:
2926 vca
= get_deployed_vca(db_nsr
, {"vdu_id": None, "member-vnf-index": None})
2928 deployed_component
= DeployedVCA(nsr_id
, vca
)
2929 elif ee_level
== EELevel
.VNF
:
2930 vca
= get_deployed_vca(
2934 "member-vnf-index": ee_relation
.vnf_profile_id
,
2935 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2939 deployed_component
= DeployedVCA(nsr_id
, vca
)
2940 elif ee_level
== EELevel
.VDU
:
2941 vca
= get_deployed_vca(
2944 "vdu_id": ee_relation
.vdu_profile_id
,
2945 "member-vnf-index": ee_relation
.vnf_profile_id
,
2946 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2950 deployed_component
= DeployedVCA(nsr_id
, vca
)
2951 elif ee_level
== EELevel
.KDU
:
2952 kdu_resource_data
= self
._get
_kdu
_resource
_data
(
2953 ee_relation
, db_nsr
, cached_vnfds
2955 if kdu_resource_data
:
2956 deployed_component
= DeployedK8sResource(kdu_resource_data
)
2957 return deployed_component
2959 async def _add_relation(
2963 db_nsr
: Dict
[str, Any
],
2964 cached_vnfds
: Dict
[str, Any
],
2965 cached_vnfrs
: Dict
[str, Any
],
2967 deployed_provider
= self
._get
_deployed
_component
(
2968 relation
.provider
, db_nsr
, cached_vnfds
2970 deployed_requirer
= self
._get
_deployed
_component
(
2971 relation
.requirer
, db_nsr
, cached_vnfds
2975 and deployed_requirer
2976 and deployed_provider
.config_sw_installed
2977 and deployed_requirer
.config_sw_installed
2979 provider_db_vnfr
= (
2981 relation
.provider
.nsr_id
,
2982 relation
.provider
.vnf_profile_id
,
2985 if relation
.provider
.vnf_profile_id
2988 requirer_db_vnfr
= (
2990 relation
.requirer
.nsr_id
,
2991 relation
.requirer
.vnf_profile_id
,
2994 if relation
.requirer
.vnf_profile_id
2997 provider_vca_id
= self
.get_vca_id(provider_db_vnfr
, db_nsr
)
2998 requirer_vca_id
= self
.get_vca_id(requirer_db_vnfr
, db_nsr
)
2999 provider_relation_endpoint
= RelationEndpoint(
3000 deployed_provider
.ee_id
,
3002 relation
.provider
.endpoint
,
3004 requirer_relation_endpoint
= RelationEndpoint(
3005 deployed_requirer
.ee_id
,
3007 relation
.requirer
.endpoint
,
3009 await self
.vca_map
[vca_type
].add_relation(
3010 provider
=provider_relation_endpoint
,
3011 requirer
=requirer_relation_endpoint
,
3013 # remove entry from relations list
3017 async def _add_vca_relations(
3023 timeout
: int = 3600,
3027 # 1. find all relations for this VCA
3028 # 2. wait for other peers related
3032 # STEP 1: find all relations for this VCA
3035 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3036 nsd
= get_nsd(db_nsr
)
3039 deployed_vca_dict
= get_deployed_vca_list(db_nsr
)[vca_index
]
3040 my_vca
= DeployedVCA(nsr_id
, deployed_vca_dict
)
3045 relations
.extend(self
._get
_ns
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
3046 relations
.extend(self
._get
_vnf
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
3048 # if no relations, terminate
3050 self
.logger
.debug(logging_text
+ " No relations")
3053 self
.logger
.debug(logging_text
+ " adding relations {}".format(relations
))
3060 if now
- start
>= timeout
:
3061 self
.logger
.error(logging_text
+ " : timeout adding relations")
3064 # reload nsr from database (we need to update record: _admin.deployed.VCA)
3065 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3067 # for each relation, find the VCA's related
3068 for relation
in relations
.copy():
3069 added
= await self
._add
_relation
(
3077 relations
.remove(relation
)
3080 self
.logger
.debug("Relations added")
3082 await asyncio
.sleep(5.0)
3086 except Exception as e
:
3087 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
3090 async def _install_kdu(
3098 k8s_instance_info
: dict,
3099 k8params
: dict = None,
3105 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
3108 "collection": "nsrs",
3109 "filter": {"_id": nsr_id
},
3110 "path": nsr_db_path
,
3113 if k8s_instance_info
.get("kdu-deployment-name"):
3114 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
3116 kdu_instance
= self
.k8scluster_map
[
3118 ].generate_kdu_instance_name(
3119 db_dict
=db_dict_install
,
3120 kdu_model
=k8s_instance_info
["kdu-model"],
3121 kdu_name
=k8s_instance_info
["kdu-name"],
3124 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3126 await self
.k8scluster_map
[k8sclustertype
].install(
3127 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3128 kdu_model
=k8s_instance_info
["kdu-model"],
3131 db_dict
=db_dict_install
,
3133 kdu_name
=k8s_instance_info
["kdu-name"],
3134 namespace
=k8s_instance_info
["namespace"],
3135 kdu_instance
=kdu_instance
,
3139 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3142 # Obtain services to obtain management service ip
3143 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
3144 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3145 kdu_instance
=kdu_instance
,
3146 namespace
=k8s_instance_info
["namespace"],
3149 # Obtain management service info (if exists)
3150 vnfr_update_dict
= {}
3151 kdu_config
= get_configuration(vnfd
, kdud
["name"])
3153 target_ee_list
= kdu_config
.get("execution-environment-list", [])
3158 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
3161 for service
in kdud
.get("service", [])
3162 if service
.get("mgmt-service")
3164 for mgmt_service
in mgmt_services
:
3165 for service
in services
:
3166 if service
["name"].startswith(mgmt_service
["name"]):
3167 # Mgmt service found, Obtain service ip
3168 ip
= service
.get("external_ip", service
.get("cluster_ip"))
3169 if isinstance(ip
, list) and len(ip
) == 1:
3173 "kdur.{}.ip-address".format(kdu_index
)
3176 # Check if must update also mgmt ip at the vnf
3177 service_external_cp
= mgmt_service
.get(
3178 "external-connection-point-ref"
3180 if service_external_cp
:
3182 deep_get(vnfd
, ("mgmt-interface", "cp"))
3183 == service_external_cp
3185 vnfr_update_dict
["ip-address"] = ip
3190 "external-connection-point-ref", ""
3192 == service_external_cp
,
3195 "kdur.{}.ip-address".format(kdu_index
)
3200 "Mgmt service name: {} not found".format(
3201 mgmt_service
["name"]
3205 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3206 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3208 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3211 and kdu_config
.get("initial-config-primitive")
3212 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3214 initial_config_primitive_list
= kdu_config
.get(
3215 "initial-config-primitive"
3217 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3219 for initial_config_primitive
in initial_config_primitive_list
:
3220 primitive_params_
= self
._map
_primitive
_params
(
3221 initial_config_primitive
, {}, {}
3224 await asyncio
.wait_for(
3225 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3226 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3227 kdu_instance
=kdu_instance
,
3228 primitive_name
=initial_config_primitive
["name"],
3229 params
=primitive_params_
,
3230 db_dict
=db_dict_install
,
3236 except Exception as e
:
3237 # Prepare update db with error and raise exception
3240 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3244 vnfr_data
.get("_id"),
3245 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3248 # ignore to keep original exception
3250 # reraise original error
3255 async def deploy_kdus(
3262 task_instantiation_info
,
3264 # Launch kdus if present in the descriptor
3266 k8scluster_id_2_uuic
= {
3267 "helm-chart-v3": {},
3272 async def _get_cluster_id(cluster_id
, cluster_type
):
3273 nonlocal k8scluster_id_2_uuic
3274 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3275 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3277 # check if K8scluster is creating and wait look if previous tasks in process
3278 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3279 "k8scluster", cluster_id
3282 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3283 task_name
, cluster_id
3285 self
.logger
.debug(logging_text
+ text
)
3286 await asyncio
.wait(task_dependency
, timeout
=3600)
3288 db_k8scluster
= self
.db
.get_one(
3289 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3291 if not db_k8scluster
:
3292 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3294 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3296 if cluster_type
== "helm-chart-v3":
3298 # backward compatibility for existing clusters that have not been initialized for helm v3
3299 k8s_credentials
= yaml
.safe_dump(
3300 db_k8scluster
.get("credentials")
3302 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3303 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3305 db_k8scluster_update
= {}
3306 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3307 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3308 db_k8scluster_update
[
3309 "_admin.helm-chart-v3.created"
3311 db_k8scluster_update
[
3312 "_admin.helm-chart-v3.operationalState"
3315 "k8sclusters", cluster_id
, db_k8scluster_update
3317 except Exception as e
:
3320 + "error initializing helm-v3 cluster: {}".format(str(e
))
3323 "K8s cluster '{}' has not been initialized for '{}'".format(
3324 cluster_id
, cluster_type
3329 "K8s cluster '{}' has not been initialized for '{}'".format(
3330 cluster_id
, cluster_type
3333 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3336 logging_text
+= "Deploy kdus: "
3339 db_nsr_update
= {"_admin.deployed.K8s": []}
3340 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3343 updated_cluster_list
= []
3344 updated_v3_cluster_list
= []
3346 for vnfr_data
in db_vnfrs
.values():
3347 vca_id
= self
.get_vca_id(vnfr_data
, {})
3348 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3349 # Step 0: Prepare and set parameters
3350 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3351 vnfd_id
= vnfr_data
.get("vnfd-id")
3352 vnfd_with_id
= find_in_list(
3353 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3357 for kdud
in vnfd_with_id
["kdu"]
3358 if kdud
["name"] == kdur
["kdu-name"]
3360 namespace
= kdur
.get("k8s-namespace")
3361 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3362 if kdur
.get("helm-chart"):
3363 kdumodel
= kdur
["helm-chart"]
3364 # Default version: helm3, if helm-version is v2 assign v2
3365 k8sclustertype
= "helm-chart-v3"
3366 self
.logger
.debug("kdur: {}".format(kdur
))
3368 kdur
.get("helm-version")
3369 and kdur
.get("helm-version") == "v2"
3371 k8sclustertype
= "helm-chart"
3372 elif kdur
.get("juju-bundle"):
3373 kdumodel
= kdur
["juju-bundle"]
3374 k8sclustertype
= "juju-bundle"
3377 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3378 "juju-bundle. Maybe an old NBI version is running".format(
3379 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3382 # check if kdumodel is a file and exists
3384 vnfd_with_id
= find_in_list(
3385 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3387 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3388 if storage
: # may be not present if vnfd has not artifacts
3389 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3390 if storage
["pkg-dir"]:
3391 filename
= "{}/{}/{}s/{}".format(
3398 filename
= "{}/Scripts/{}s/{}".format(
3403 if self
.fs
.file_exists(
3404 filename
, mode
="file"
3405 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3406 kdumodel
= self
.fs
.path
+ filename
3407 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3409 except Exception: # it is not a file
3412 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3413 step
= "Synchronize repos for k8s cluster '{}'".format(
3416 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3420 k8sclustertype
== "helm-chart"
3421 and cluster_uuid
not in updated_cluster_list
3423 k8sclustertype
== "helm-chart-v3"
3424 and cluster_uuid
not in updated_v3_cluster_list
3426 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3427 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3428 cluster_uuid
=cluster_uuid
3431 if del_repo_list
or added_repo_dict
:
3432 if k8sclustertype
== "helm-chart":
3434 "_admin.helm_charts_added." + item
: None
3435 for item
in del_repo_list
3438 "_admin.helm_charts_added." + item
: name
3439 for item
, name
in added_repo_dict
.items()
3441 updated_cluster_list
.append(cluster_uuid
)
3442 elif k8sclustertype
== "helm-chart-v3":
3444 "_admin.helm_charts_v3_added." + item
: None
3445 for item
in del_repo_list
3448 "_admin.helm_charts_v3_added." + item
: name
3449 for item
, name
in added_repo_dict
.items()
3451 updated_v3_cluster_list
.append(cluster_uuid
)
3453 logging_text
+ "repos synchronized on k8s cluster "
3454 "'{}' to_delete: {}, to_add: {}".format(
3455 k8s_cluster_id
, del_repo_list
, added_repo_dict
3460 {"_id": k8s_cluster_id
},
3466 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3467 vnfr_data
["member-vnf-index-ref"],
3471 k8s_instance_info
= {
3472 "kdu-instance": None,
3473 "k8scluster-uuid": cluster_uuid
,
3474 "k8scluster-type": k8sclustertype
,
3475 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3476 "kdu-name": kdur
["kdu-name"],
3477 "kdu-model": kdumodel
,
3478 "namespace": namespace
,
3479 "kdu-deployment-name": kdu_deployment_name
,
3481 db_path
= "_admin.deployed.K8s.{}".format(index
)
3482 db_nsr_update
[db_path
] = k8s_instance_info
3483 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3484 vnfd_with_id
= find_in_list(
3485 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3487 task
= asyncio
.ensure_future(
3496 k8params
=desc_params
,
3501 self
.lcm_tasks
.register(
3505 "instantiate_KDU-{}".format(index
),
3508 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3514 except (LcmException
, asyncio
.CancelledError
):
3516 except Exception as e
:
3517 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3518 if isinstance(e
, (N2VCException
, DbException
)):
3519 self
.logger
.error(logging_text
+ msg
)
3521 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3522 raise LcmException(msg
)
3525 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3544 task_instantiation_info
,
3547 # launch instantiate_N2VC in a asyncio task and register task object
3548 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3549 # if not found, create one entry and update database
3550 # fill db_nsr._admin.deployed.VCA.<index>
3553 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3555 if "execution-environment-list" in descriptor_config
:
3556 ee_list
= descriptor_config
.get("execution-environment-list", [])
3557 elif "juju" in descriptor_config
:
3558 ee_list
= [descriptor_config
] # ns charms
3559 else: # other types as script are not supported
3562 for ee_item
in ee_list
:
3565 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3566 ee_item
.get("juju"), ee_item
.get("helm-chart")
3569 ee_descriptor_id
= ee_item
.get("id")
3570 if ee_item
.get("juju"):
3571 vca_name
= ee_item
["juju"].get("charm")
3574 if ee_item
["juju"].get("charm") is not None
3577 if ee_item
["juju"].get("cloud") == "k8s":
3578 vca_type
= "k8s_proxy_charm"
3579 elif ee_item
["juju"].get("proxy") is False:
3580 vca_type
= "native_charm"
3581 elif ee_item
.get("helm-chart"):
3582 vca_name
= ee_item
["helm-chart"]
3583 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3586 vca_type
= "helm-v3"
3589 logging_text
+ "skipping non juju neither charm configuration"
3594 for vca_index
, vca_deployed
in enumerate(
3595 db_nsr
["_admin"]["deployed"]["VCA"]
3597 if not vca_deployed
:
3600 vca_deployed
.get("member-vnf-index") == member_vnf_index
3601 and vca_deployed
.get("vdu_id") == vdu_id
3602 and vca_deployed
.get("kdu_name") == kdu_name
3603 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3604 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3608 # not found, create one.
3610 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3613 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3615 target
+= "/kdu/{}".format(kdu_name
)
3617 "target_element": target
,
3618 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3619 "member-vnf-index": member_vnf_index
,
3621 "kdu_name": kdu_name
,
3622 "vdu_count_index": vdu_index
,
3623 "operational-status": "init", # TODO revise
3624 "detailed-status": "", # TODO revise
3625 "step": "initial-deploy", # TODO revise
3627 "vdu_name": vdu_name
,
3629 "ee_descriptor_id": ee_descriptor_id
,
3633 # create VCA and configurationStatus in db
3635 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3636 "configurationStatus.{}".format(vca_index
): dict(),
3638 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3640 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3642 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3643 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3644 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3647 task_n2vc
= asyncio
.ensure_future(
3648 self
.instantiate_N2VC(
3649 logging_text
=logging_text
,
3650 vca_index
=vca_index
,
3656 vdu_index
=vdu_index
,
3657 deploy_params
=deploy_params
,
3658 config_descriptor
=descriptor_config
,
3659 base_folder
=base_folder
,
3660 nslcmop_id
=nslcmop_id
,
3664 ee_config_descriptor
=ee_item
,
3667 self
.lcm_tasks
.register(
3671 "instantiate_N2VC-{}".format(vca_index
),
3674 task_instantiation_info
[
3676 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3677 member_vnf_index
or "", vdu_id
or ""
3681 def _create_nslcmop(nsr_id
, operation
, params
):
3683 Creates a ns-lcm-opp content to be stored at database.
3684 :param nsr_id: internal id of the instance
3685 :param operation: instantiate, terminate, scale, action, ...
3686 :param params: user parameters for the operation
3687 :return: dictionary following SOL005 format
3689 # Raise exception if invalid arguments
3690 if not (nsr_id
and operation
and params
):
3692 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3699 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3700 "operationState": "PROCESSING",
3701 "statusEnteredTime": now
,
3702 "nsInstanceId": nsr_id
,
3703 "lcmOperationType": operation
,
3705 "isAutomaticInvocation": False,
3706 "operationParams": params
,
3707 "isCancelPending": False,
3709 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3710 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3715 def _format_additional_params(self
, params
):
3716 params
= params
or {}
3717 for key
, value
in params
.items():
3718 if str(value
).startswith("!!yaml "):
3719 params
[key
] = yaml
.safe_load(value
[7:])
3722 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3723 primitive
= seq
.get("name")
3724 primitive_params
= {}
3726 "member_vnf_index": vnf_index
,
3727 "primitive": primitive
,
3728 "primitive_params": primitive_params
,
3731 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3735 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3736 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3737 if op
.get("operationState") == "COMPLETED":
3738 # b. Skip sub-operation
3739 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3740 return self
.SUBOPERATION_STATUS_SKIP
3742 # c. retry executing sub-operation
3743 # The sub-operation exists, and operationState != 'COMPLETED'
3744 # Update operationState = 'PROCESSING' to indicate a retry.
3745 operationState
= "PROCESSING"
3746 detailed_status
= "In progress"
3747 self
._update
_suboperation
_status
(
3748 db_nslcmop
, op_index
, operationState
, detailed_status
3750 # Return the sub-operation index
3751 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3752 # with arguments extracted from the sub-operation
3755 # Find a sub-operation where all keys in a matching dictionary must match
3756 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3757 def _find_suboperation(self
, db_nslcmop
, match
):
3758 if db_nslcmop
and match
:
3759 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3760 for i
, op
in enumerate(op_list
):
3761 if all(op
.get(k
) == match
[k
] for k
in match
):
3763 return self
.SUBOPERATION_STATUS_NOT_FOUND
3765 # Update status for a sub-operation given its index
3766 def _update_suboperation_status(
3767 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3769 # Update DB for HA tasks
3770 q_filter
= {"_id": db_nslcmop
["_id"]}
3772 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3773 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3776 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3779 # Add sub-operation, return the index of the added sub-operation
3780 # Optionally, set operationState, detailed-status, and operationType
3781 # Status and type are currently set for 'scale' sub-operations:
3782 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3783 # 'detailed-status' : status message
3784 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3785 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3786 def _add_suboperation(
3794 mapped_primitive_params
,
3795 operationState
=None,
3796 detailed_status
=None,
3799 RO_scaling_info
=None,
3802 return self
.SUBOPERATION_STATUS_NOT_FOUND
3803 # Get the "_admin.operations" list, if it exists
3804 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3805 op_list
= db_nslcmop_admin
.get("operations")
3806 # Create or append to the "_admin.operations" list
3808 "member_vnf_index": vnf_index
,
3810 "vdu_count_index": vdu_count_index
,
3811 "primitive": primitive
,
3812 "primitive_params": mapped_primitive_params
,
3815 new_op
["operationState"] = operationState
3817 new_op
["detailed-status"] = detailed_status
3819 new_op
["lcmOperationType"] = operationType
3821 new_op
["RO_nsr_id"] = RO_nsr_id
3823 new_op
["RO_scaling_info"] = RO_scaling_info
3825 # No existing operations, create key 'operations' with current operation as first list element
3826 db_nslcmop_admin
.update({"operations": [new_op
]})
3827 op_list
= db_nslcmop_admin
.get("operations")
3829 # Existing operations, append operation to list
3830 op_list
.append(new_op
)
3832 db_nslcmop_update
= {"_admin.operations": op_list
}
3833 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3834 op_index
= len(op_list
) - 1
3837 # Helper methods for scale() sub-operations
3839 # pre-scale/post-scale:
3840 # Check for 3 different cases:
3841 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3842 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3843 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3844 def _check_or_add_scale_suboperation(
3848 vnf_config_primitive
,
3852 RO_scaling_info
=None,
3854 # Find this sub-operation
3855 if RO_nsr_id
and RO_scaling_info
:
3856 operationType
= "SCALE-RO"
3858 "member_vnf_index": vnf_index
,
3859 "RO_nsr_id": RO_nsr_id
,
3860 "RO_scaling_info": RO_scaling_info
,
3864 "member_vnf_index": vnf_index
,
3865 "primitive": vnf_config_primitive
,
3866 "primitive_params": primitive_params
,
3867 "lcmOperationType": operationType
,
3869 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3870 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3871 # a. New sub-operation
3872 # The sub-operation does not exist, add it.
3873 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3874 # The following parameters are set to None for all kind of scaling:
3876 vdu_count_index
= None
3878 if RO_nsr_id
and RO_scaling_info
:
3879 vnf_config_primitive
= None
3880 primitive_params
= None
3883 RO_scaling_info
= None
3884 # Initial status for sub-operation
3885 operationState
= "PROCESSING"
3886 detailed_status
= "In progress"
3887 # Add sub-operation for pre/post-scaling (zero or more operations)
3888 self
._add
_suboperation
(
3894 vnf_config_primitive
,
3902 return self
.SUBOPERATION_STATUS_NEW
3904 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3905 # or op_index (operationState != 'COMPLETED')
3906 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3908 # Function to return execution_environment id
3910 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3911 # TODO vdu_index_count
3912 for vca
in vca_deployed_list
:
3913 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3916 async def destroy_N2VC(
3924 exec_primitives
=True,
3929 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3930 :param logging_text:
3932 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3933 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3934 :param vca_index: index in the database _admin.deployed.VCA
3935 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3936 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3937 not executed properly
3938 :param scaling_in: True destroys the application, False destroys the model
3939 :return: None or exception
3944 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3945 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3949 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3951 # execute terminate_primitives
3953 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3954 config_descriptor
.get("terminate-config-primitive"),
3955 vca_deployed
.get("ee_descriptor_id"),
3957 vdu_id
= vca_deployed
.get("vdu_id")
3958 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3959 vdu_name
= vca_deployed
.get("vdu_name")
3960 vnf_index
= vca_deployed
.get("member-vnf-index")
3961 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3962 for seq
in terminate_primitives
:
3963 # For each sequence in list, get primitive and call _ns_execute_primitive()
3964 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3965 vnf_index
, seq
.get("name")
3967 self
.logger
.debug(logging_text
+ step
)
3968 # Create the primitive for each sequence, i.e. "primitive": "touch"
3969 primitive
= seq
.get("name")
3970 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3975 self
._add
_suboperation
(
3982 mapped_primitive_params
,
3984 # Sub-operations: Call _ns_execute_primitive() instead of action()
3986 result
, result_detail
= await self
._ns
_execute
_primitive
(
3987 vca_deployed
["ee_id"],
3989 mapped_primitive_params
,
3993 except LcmException
:
3994 # this happens when VCA is not deployed. In this case it is not needed to terminate
3996 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3997 if result
not in result_ok
:
3999 "terminate_primitive {} for vnf_member_index={} fails with "
4000 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
4002 # set that this VCA do not need terminated
4003 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
4007 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
4010 # Delete Prometheus Jobs if any
4011 # This uses NSR_ID, so it will destroy any jobs under this index
4012 self
.db
.del_list("prometheus_jobs", {"nsr_id": db_nslcmop
["nsInstanceId"]})
4015 await self
.vca_map
[vca_type
].delete_execution_environment(
4016 vca_deployed
["ee_id"],
4017 scaling_in
=scaling_in
,
4022 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
4023 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
4024 namespace
= "." + db_nsr
["_id"]
4026 await self
.n2vc
.delete_namespace(
4027 namespace
=namespace
,
4028 total_timeout
=self
.timeout_charm_delete
,
4031 except N2VCNotFound
: # already deleted. Skip
4033 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
4035 async def _terminate_RO(
4036 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4039 Terminates a deployment from RO
4040 :param logging_text:
4041 :param nsr_deployed: db_nsr._admin.deployed
4044 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
4045 this method will update only the index 2, but it will write on database the concatenated content of the list
4050 ro_nsr_id
= ro_delete_action
= None
4051 if nsr_deployed
and nsr_deployed
.get("RO"):
4052 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
4053 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
4056 stage
[2] = "Deleting ns from VIM."
4057 db_nsr_update
["detailed-status"] = " ".join(stage
)
4058 self
._write
_op
_status
(nslcmop_id
, stage
)
4059 self
.logger
.debug(logging_text
+ stage
[2])
4060 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4061 self
._write
_op
_status
(nslcmop_id
, stage
)
4062 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
4063 ro_delete_action
= desc
["action_id"]
4065 "_admin.deployed.RO.nsr_delete_action_id"
4066 ] = ro_delete_action
4067 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4068 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4069 if ro_delete_action
:
4070 # wait until NS is deleted from VIM
4071 stage
[2] = "Waiting ns deleted from VIM."
4072 detailed_status_old
= None
4076 + " RO_id={} ro_delete_action={}".format(
4077 ro_nsr_id
, ro_delete_action
4080 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4081 self
._write
_op
_status
(nslcmop_id
, stage
)
4083 delete_timeout
= 20 * 60 # 20 minutes
4084 while delete_timeout
> 0:
4085 desc
= await self
.RO
.show(
4087 item_id_name
=ro_nsr_id
,
4088 extra_item
="action",
4089 extra_item_id
=ro_delete_action
,
4093 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4095 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4096 if ns_status
== "ERROR":
4097 raise ROclient
.ROClientException(ns_status_info
)
4098 elif ns_status
== "BUILD":
4099 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
4100 elif ns_status
== "ACTIVE":
4101 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4102 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4107 ), "ROclient.check_action_status returns unknown {}".format(
4110 if stage
[2] != detailed_status_old
:
4111 detailed_status_old
= stage
[2]
4112 db_nsr_update
["detailed-status"] = " ".join(stage
)
4113 self
._write
_op
_status
(nslcmop_id
, stage
)
4114 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4115 await asyncio
.sleep(5, loop
=self
.loop
)
4117 else: # delete_timeout <= 0:
4118 raise ROclient
.ROClientException(
4119 "Timeout waiting ns deleted from VIM"
4122 except Exception as e
:
4123 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4125 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4127 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4128 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4129 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4131 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
4134 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4136 failed_detail
.append("delete conflict: {}".format(e
))
4139 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
4142 failed_detail
.append("delete error: {}".format(e
))
4144 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
4148 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
4149 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
4151 stage
[2] = "Deleting nsd from RO."
4152 db_nsr_update
["detailed-status"] = " ".join(stage
)
4153 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4154 self
._write
_op
_status
(nslcmop_id
, stage
)
4155 await self
.RO
.delete("nsd", ro_nsd_id
)
4157 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
4159 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4160 except Exception as e
:
4162 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4164 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4166 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
4169 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4171 failed_detail
.append(
4172 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
4174 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4176 failed_detail
.append(
4177 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
4179 self
.logger
.error(logging_text
+ failed_detail
[-1])
4181 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
4182 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
4183 if not vnf_deployed
or not vnf_deployed
["id"]:
4186 ro_vnfd_id
= vnf_deployed
["id"]
4189 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
4190 vnf_deployed
["member-vnf-index"], ro_vnfd_id
4192 db_nsr_update
["detailed-status"] = " ".join(stage
)
4193 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4194 self
._write
_op
_status
(nslcmop_id
, stage
)
4195 await self
.RO
.delete("vnfd", ro_vnfd_id
)
4197 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
4199 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
4200 except Exception as e
:
4202 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4205 "_admin.deployed.RO.vnfd.{}.id".format(index
)
4209 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4212 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4214 failed_detail
.append(
4215 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4217 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4219 failed_detail
.append(
4220 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4222 self
.logger
.error(logging_text
+ failed_detail
[-1])
4225 stage
[2] = "Error deleting from VIM"
4227 stage
[2] = "Deleted from VIM"
4228 db_nsr_update
["detailed-status"] = " ".join(stage
)
4229 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4230 self
._write
_op
_status
(nslcmop_id
, stage
)
4233 raise LcmException("; ".join(failed_detail
))
4235 async def terminate(self
, nsr_id
, nslcmop_id
):
4236 # Try to lock HA task here
4237 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4238 if not task_is_locked_by_me
:
4241 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4242 self
.logger
.debug(logging_text
+ "Enter")
4243 timeout_ns_terminate
= self
.timeout_ns_terminate
4246 operation_params
= None
4248 error_list
= [] # annotates all failed error messages
4249 db_nslcmop_update
= {}
4250 autoremove
= False # autoremove after terminated
4251 tasks_dict_info
= {}
4254 "Stage 1/3: Preparing task.",
4255 "Waiting for previous operations to terminate.",
4258 # ^ contains [stage, step, VIM-status]
4260 # wait for any previous tasks in process
4261 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4263 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4264 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4265 operation_params
= db_nslcmop
.get("operationParams") or {}
4266 if operation_params
.get("timeout_ns_terminate"):
4267 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4268 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4269 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4271 db_nsr_update
["operational-status"] = "terminating"
4272 db_nsr_update
["config-status"] = "terminating"
4273 self
._write
_ns
_status
(
4275 ns_state
="TERMINATING",
4276 current_operation
="TERMINATING",
4277 current_operation_id
=nslcmop_id
,
4278 other_update
=db_nsr_update
,
4280 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4281 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4282 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4285 stage
[1] = "Getting vnf descriptors from db."
4286 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4288 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4290 db_vnfds_from_id
= {}
4291 db_vnfds_from_member_index
= {}
4293 for vnfr
in db_vnfrs_list
:
4294 vnfd_id
= vnfr
["vnfd-id"]
4295 if vnfd_id
not in db_vnfds_from_id
:
4296 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4297 db_vnfds_from_id
[vnfd_id
] = vnfd
4298 db_vnfds_from_member_index
[
4299 vnfr
["member-vnf-index-ref"]
4300 ] = db_vnfds_from_id
[vnfd_id
]
4302 # Destroy individual execution environments when there are terminating primitives.
4303 # Rest of EE will be deleted at once
4304 # TODO - check before calling _destroy_N2VC
4305 # if not operation_params.get("skip_terminate_primitives"):#
4306 # or not vca.get("needed_terminate"):
4307 stage
[0] = "Stage 2/3 execute terminating primitives."
4308 self
.logger
.debug(logging_text
+ stage
[0])
4309 stage
[1] = "Looking execution environment that needs terminate."
4310 self
.logger
.debug(logging_text
+ stage
[1])
4312 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4313 config_descriptor
= None
4314 vca_member_vnf_index
= vca
.get("member-vnf-index")
4315 vca_id
= self
.get_vca_id(
4316 db_vnfrs_dict
.get(vca_member_vnf_index
)
4317 if vca_member_vnf_index
4321 if not vca
or not vca
.get("ee_id"):
4323 if not vca
.get("member-vnf-index"):
4325 config_descriptor
= db_nsr
.get("ns-configuration")
4326 elif vca
.get("vdu_id"):
4327 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4328 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4329 elif vca
.get("kdu_name"):
4330 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4331 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4333 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4334 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4335 vca_type
= vca
.get("type")
4336 exec_terminate_primitives
= not operation_params
.get(
4337 "skip_terminate_primitives"
4338 ) and vca
.get("needed_terminate")
4339 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4340 # pending native charms
4342 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4344 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4345 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4346 task
= asyncio
.ensure_future(
4354 exec_terminate_primitives
,
4358 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4360 # wait for pending tasks of terminate primitives
4364 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4366 error_list
= await self
._wait
_for
_tasks
(
4369 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4373 tasks_dict_info
.clear()
4375 return # raise LcmException("; ".join(error_list))
4377 # remove All execution environments at once
4378 stage
[0] = "Stage 3/3 delete all."
4380 if nsr_deployed
.get("VCA"):
4381 stage
[1] = "Deleting all execution environments."
4382 self
.logger
.debug(logging_text
+ stage
[1])
4383 vca_id
= self
.get_vca_id({}, db_nsr
)
4384 task_delete_ee
= asyncio
.ensure_future(
4386 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4387 timeout
=self
.timeout_charm_delete
,
4390 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4391 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4393 # Delete from k8scluster
4394 stage
[1] = "Deleting KDUs."
4395 self
.logger
.debug(logging_text
+ stage
[1])
4396 # print(nsr_deployed)
4397 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4398 if not kdu
or not kdu
.get("kdu-instance"):
4400 kdu_instance
= kdu
.get("kdu-instance")
4401 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4402 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4403 vca_id
= self
.get_vca_id({}, db_nsr
)
4404 task_delete_kdu_instance
= asyncio
.ensure_future(
4405 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4406 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4407 kdu_instance
=kdu_instance
,
4414 + "Unknown k8s deployment type {}".format(
4415 kdu
.get("k8scluster-type")
4420 task_delete_kdu_instance
4421 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4424 stage
[1] = "Deleting ns from VIM."
4426 task_delete_ro
= asyncio
.ensure_future(
4427 self
._terminate
_ng
_ro
(
4428 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4432 task_delete_ro
= asyncio
.ensure_future(
4434 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4437 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4439 # rest of staff will be done at finally
4442 ROclient
.ROClientException
,
4447 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4449 except asyncio
.CancelledError
:
4451 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4453 exc
= "Operation was cancelled"
4454 except Exception as e
:
4455 exc
= traceback
.format_exc()
4456 self
.logger
.critical(
4457 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4462 error_list
.append(str(exc
))
4464 # wait for pending tasks
4466 stage
[1] = "Waiting for terminate pending tasks."
4467 self
.logger
.debug(logging_text
+ stage
[1])
4468 error_list
+= await self
._wait
_for
_tasks
(
4471 timeout_ns_terminate
,
4475 stage
[1] = stage
[2] = ""
4476 except asyncio
.CancelledError
:
4477 error_list
.append("Cancelled")
4478 # TODO cancell all tasks
4479 except Exception as exc
:
4480 error_list
.append(str(exc
))
4481 # update status at database
4483 error_detail
= "; ".join(error_list
)
4484 # self.logger.error(logging_text + error_detail)
4485 error_description_nslcmop
= "{} Detail: {}".format(
4486 stage
[0], error_detail
4488 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4489 nslcmop_id
, stage
[0]
4492 db_nsr_update
["operational-status"] = "failed"
4493 db_nsr_update
["detailed-status"] = (
4494 error_description_nsr
+ " Detail: " + error_detail
4496 db_nslcmop_update
["detailed-status"] = error_detail
4497 nslcmop_operation_state
= "FAILED"
4501 error_description_nsr
= error_description_nslcmop
= None
4502 ns_state
= "NOT_INSTANTIATED"
4503 db_nsr_update
["operational-status"] = "terminated"
4504 db_nsr_update
["detailed-status"] = "Done"
4505 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4506 db_nslcmop_update
["detailed-status"] = "Done"
4507 nslcmop_operation_state
= "COMPLETED"
4510 self
._write
_ns
_status
(
4513 current_operation
="IDLE",
4514 current_operation_id
=None,
4515 error_description
=error_description_nsr
,
4516 error_detail
=error_detail
,
4517 other_update
=db_nsr_update
,
4519 self
._write
_op
_status
(
4522 error_message
=error_description_nslcmop
,
4523 operation_state
=nslcmop_operation_state
,
4524 other_update
=db_nslcmop_update
,
4526 if ns_state
== "NOT_INSTANTIATED":
4530 {"nsr-id-ref": nsr_id
},
4531 {"_admin.nsState": "NOT_INSTANTIATED"},
4533 except DbException
as e
:
4536 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4540 if operation_params
:
4541 autoremove
= operation_params
.get("autoremove", False)
4542 if nslcmop_operation_state
:
4544 await self
.msg
.aiowrite(
4549 "nslcmop_id": nslcmop_id
,
4550 "operationState": nslcmop_operation_state
,
4551 "autoremove": autoremove
,
4555 except Exception as e
:
4557 logging_text
+ "kafka_write notification Exception {}".format(e
)
4560 self
.logger
.debug(logging_text
+ "Exit")
4561 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4563 async def _wait_for_tasks(
4564 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4567 error_detail_list
= []
4569 pending_tasks
= list(created_tasks_info
.keys())
4570 num_tasks
= len(pending_tasks
)
4572 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4573 self
._write
_op
_status
(nslcmop_id
, stage
)
4574 while pending_tasks
:
4576 _timeout
= timeout
+ time_start
- time()
4577 done
, pending_tasks
= await asyncio
.wait(
4578 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4580 num_done
+= len(done
)
4581 if not done
: # Timeout
4582 for task
in pending_tasks
:
4583 new_error
= created_tasks_info
[task
] + ": Timeout"
4584 error_detail_list
.append(new_error
)
4585 error_list
.append(new_error
)
4588 if task
.cancelled():
4591 exc
= task
.exception()
4593 if isinstance(exc
, asyncio
.TimeoutError
):
4595 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4596 error_list
.append(created_tasks_info
[task
])
4597 error_detail_list
.append(new_error
)
4604 ROclient
.ROClientException
,
4610 self
.logger
.error(logging_text
+ new_error
)
4612 exc_traceback
= "".join(
4613 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4617 + created_tasks_info
[task
]
4623 logging_text
+ created_tasks_info
[task
] + ": Done"
4625 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4627 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4628 if nsr_id
: # update also nsr
4633 "errorDescription": "Error at: " + ", ".join(error_list
),
4634 "errorDetail": ". ".join(error_detail_list
),
4637 self
._write
_op
_status
(nslcmop_id
, stage
)
4638 return error_detail_list
4641 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4643 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4644 The default-value is used. If it is between < > it look for a value at instantiation_params
4645 :param primitive_desc: portion of VNFD/NSD that describes primitive
4646 :param params: Params provided by user
4647 :param instantiation_params: Instantiation params provided by user
4648 :return: a dictionary with the calculated params
4650 calculated_params
= {}
4651 for parameter
in primitive_desc
.get("parameter", ()):
4652 param_name
= parameter
["name"]
4653 if param_name
in params
:
4654 calculated_params
[param_name
] = params
[param_name
]
4655 elif "default-value" in parameter
or "value" in parameter
:
4656 if "value" in parameter
:
4657 calculated_params
[param_name
] = parameter
["value"]
4659 calculated_params
[param_name
] = parameter
["default-value"]
4661 isinstance(calculated_params
[param_name
], str)
4662 and calculated_params
[param_name
].startswith("<")
4663 and calculated_params
[param_name
].endswith(">")
4665 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4666 calculated_params
[param_name
] = instantiation_params
[
4667 calculated_params
[param_name
][1:-1]
4671 "Parameter {} needed to execute primitive {} not provided".format(
4672 calculated_params
[param_name
], primitive_desc
["name"]
4677 "Parameter {} needed to execute primitive {} not provided".format(
4678 param_name
, primitive_desc
["name"]
4682 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4683 calculated_params
[param_name
] = yaml
.safe_dump(
4684 calculated_params
[param_name
], default_flow_style
=True, width
=256
4686 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4688 ].startswith("!!yaml "):
4689 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4690 if parameter
.get("data-type") == "INTEGER":
4692 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4693 except ValueError: # error converting string to int
4695 "Parameter {} of primitive {} must be integer".format(
4696 param_name
, primitive_desc
["name"]
4699 elif parameter
.get("data-type") == "BOOLEAN":
4700 calculated_params
[param_name
] = not (
4701 (str(calculated_params
[param_name
])).lower() == "false"
4704 # add always ns_config_info if primitive name is config
4705 if primitive_desc
["name"] == "config":
4706 if "ns_config_info" in instantiation_params
:
4707 calculated_params
["ns_config_info"] = instantiation_params
[
4710 return calculated_params
4712 def _look_for_deployed_vca(
4719 ee_descriptor_id
=None,
4721 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4722 for vca
in deployed_vca
:
4725 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4728 vdu_count_index
is not None
4729 and vdu_count_index
!= vca
["vdu_count_index"]
4732 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4734 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4738 # vca_deployed not found
4740 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4741 " is not deployed".format(
4750 ee_id
= vca
.get("ee_id")
4752 "type", "lxc_proxy_charm"
4753 ) # default value for backward compatibility - proxy charm
4756 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4757 "execution environment".format(
4758 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4761 return ee_id
, vca_type
4763 async def _ns_execute_primitive(
4769 retries_interval
=30,
4776 if primitive
== "config":
4777 primitive_params
= {"params": primitive_params
}
4779 vca_type
= vca_type
or "lxc_proxy_charm"
4783 output
= await asyncio
.wait_for(
4784 self
.vca_map
[vca_type
].exec_primitive(
4786 primitive_name
=primitive
,
4787 params_dict
=primitive_params
,
4788 progress_timeout
=self
.timeout_progress_primitive
,
4789 total_timeout
=self
.timeout_primitive
,
4794 timeout
=timeout
or self
.timeout_primitive
,
4798 except asyncio
.CancelledError
:
4800 except Exception as e
: # asyncio.TimeoutError
4801 if isinstance(e
, asyncio
.TimeoutError
):
4806 "Error executing action {} on {} -> {}".format(
4811 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4813 return "FAILED", str(e
)
4815 return "COMPLETED", output
4817 except (LcmException
, asyncio
.CancelledError
):
4819 except Exception as e
:
4820 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4822 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4824 Updating the vca_status with latest juju information in nsrs record
4825 :param: nsr_id: Id of the nsr
4826 :param: nslcmop_id: Id of the nslcmop
4830 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4831 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4832 vca_id
= self
.get_vca_id({}, db_nsr
)
4833 if db_nsr
["_admin"]["deployed"]["K8s"]:
4834 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4835 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4836 await self
._on
_update
_k
8s
_db
(
4837 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4840 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4841 table
, filter = "nsrs", {"_id": nsr_id
}
4842 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4843 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4845 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4846 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4848 async def action(self
, nsr_id
, nslcmop_id
):
4849 # Try to lock HA task here
4850 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4851 if not task_is_locked_by_me
:
4854 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4855 self
.logger
.debug(logging_text
+ "Enter")
4856 # get all needed from database
4860 db_nslcmop_update
= {}
4861 nslcmop_operation_state
= None
4862 error_description_nslcmop
= None
4865 # wait for any previous tasks in process
4866 step
= "Waiting for previous operations to terminate"
4867 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4869 self
._write
_ns
_status
(
4872 current_operation
="RUNNING ACTION",
4873 current_operation_id
=nslcmop_id
,
4876 step
= "Getting information from database"
4877 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4878 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4879 if db_nslcmop
["operationParams"].get("primitive_params"):
4880 db_nslcmop
["operationParams"]["primitive_params"] = json
.loads(
4881 db_nslcmop
["operationParams"]["primitive_params"]
4884 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4885 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4886 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4887 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4888 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4889 primitive
= db_nslcmop
["operationParams"]["primitive"]
4890 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4891 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4892 "timeout_ns_action", self
.timeout_primitive
4896 step
= "Getting vnfr from database"
4897 db_vnfr
= self
.db
.get_one(
4898 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4900 if db_vnfr
.get("kdur"):
4902 for kdur
in db_vnfr
["kdur"]:
4903 if kdur
.get("additionalParams"):
4904 kdur
["additionalParams"] = json
.loads(kdur
["additionalParams"])
4905 kdur_list
.append(kdur
)
4906 db_vnfr
["kdur"] = kdur_list
4907 step
= "Getting vnfd from database"
4908 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4910 # Sync filesystem before running a primitive
4911 self
.fs
.sync(db_vnfr
["vnfd-id"])
4913 step
= "Getting nsd from database"
4914 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4916 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4917 # for backward compatibility
4918 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4919 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4920 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4921 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4923 # look for primitive
4924 config_primitive_desc
= descriptor_configuration
= None
4926 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4928 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4930 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4932 descriptor_configuration
= db_nsd
.get("ns-configuration")
4934 if descriptor_configuration
and descriptor_configuration
.get(
4937 for config_primitive
in descriptor_configuration
["config-primitive"]:
4938 if config_primitive
["name"] == primitive
:
4939 config_primitive_desc
= config_primitive
4942 if not config_primitive_desc
:
4943 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4945 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4949 primitive_name
= primitive
4950 ee_descriptor_id
= None
4952 primitive_name
= config_primitive_desc
.get(
4953 "execution-environment-primitive", primitive
4955 ee_descriptor_id
= config_primitive_desc
.get(
4956 "execution-environment-ref"
4962 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4964 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4967 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4969 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4971 desc_params
= parse_yaml_strings(
4972 db_vnfr
.get("additionalParamsForVnf")
4975 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4976 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4977 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4979 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4980 actions
.add(primitive
["name"])
4981 for primitive
in kdu_configuration
.get("config-primitive", []):
4982 actions
.add(primitive
["name"])
4983 kdu_action
= True if primitive_name
in actions
else False
4985 # TODO check if ns is in a proper status
4987 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4989 # kdur and desc_params already set from before
4990 if primitive_params
:
4991 desc_params
.update(primitive_params
)
4992 # TODO Check if we will need something at vnf level
4993 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4995 kdu_name
== kdu
["kdu-name"]
4996 and kdu
["member-vnf-index"] == vnf_index
5001 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
5004 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
5005 msg
= "unknown k8scluster-type '{}'".format(
5006 kdu
.get("k8scluster-type")
5008 raise LcmException(msg
)
5011 "collection": "nsrs",
5012 "filter": {"_id": nsr_id
},
5013 "path": "_admin.deployed.K8s.{}".format(index
),
5017 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
5019 step
= "Executing kdu {}".format(primitive_name
)
5020 if primitive_name
== "upgrade":
5021 if desc_params
.get("kdu_model"):
5022 kdu_model
= desc_params
.get("kdu_model")
5023 del desc_params
["kdu_model"]
5025 kdu_model
= kdu
.get("kdu-model")
5026 parts
= kdu_model
.split(sep
=":")
5028 kdu_model
= parts
[0]
5030 detailed_status
= await asyncio
.wait_for(
5031 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
5032 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5033 kdu_instance
=kdu
.get("kdu-instance"),
5035 kdu_model
=kdu_model
,
5038 timeout
=timeout_ns_action
,
5040 timeout
=timeout_ns_action
+ 10,
5043 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
5045 elif primitive_name
== "rollback":
5046 detailed_status
= await asyncio
.wait_for(
5047 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
5048 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5049 kdu_instance
=kdu
.get("kdu-instance"),
5052 timeout
=timeout_ns_action
,
5054 elif primitive_name
== "status":
5055 detailed_status
= await asyncio
.wait_for(
5056 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
5057 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5058 kdu_instance
=kdu
.get("kdu-instance"),
5061 timeout
=timeout_ns_action
,
5064 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
5065 kdu
["kdu-name"], nsr_id
5067 params
= self
._map
_primitive
_params
(
5068 config_primitive_desc
, primitive_params
, desc_params
5071 detailed_status
= await asyncio
.wait_for(
5072 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
5073 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5074 kdu_instance
=kdu_instance
,
5075 primitive_name
=primitive_name
,
5078 timeout
=timeout_ns_action
,
5081 timeout
=timeout_ns_action
,
5085 nslcmop_operation_state
= "COMPLETED"
5087 detailed_status
= ""
5088 nslcmop_operation_state
= "FAILED"
5090 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5091 nsr_deployed
["VCA"],
5092 member_vnf_index
=vnf_index
,
5094 vdu_count_index
=vdu_count_index
,
5095 ee_descriptor_id
=ee_descriptor_id
,
5097 for vca_index
, vca_deployed
in enumerate(
5098 db_nsr
["_admin"]["deployed"]["VCA"]
5100 if vca_deployed
.get("member-vnf-index") == vnf_index
:
5102 "collection": "nsrs",
5103 "filter": {"_id": nsr_id
},
5104 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
5108 nslcmop_operation_state
,
5110 ) = await self
._ns
_execute
_primitive
(
5112 primitive
=primitive_name
,
5113 primitive_params
=self
._map
_primitive
_params
(
5114 config_primitive_desc
, primitive_params
, desc_params
5116 timeout
=timeout_ns_action
,
5122 db_nslcmop_update
["detailed-status"] = detailed_status
5123 error_description_nslcmop
= (
5124 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
5128 + " task Done with result {} {}".format(
5129 nslcmop_operation_state
, detailed_status
5132 return # database update is called inside finally
5134 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5135 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5137 except asyncio
.CancelledError
:
5139 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5141 exc
= "Operation was cancelled"
5142 except asyncio
.TimeoutError
:
5143 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5145 except Exception as e
:
5146 exc
= traceback
.format_exc()
5147 self
.logger
.critical(
5148 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5157 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5158 nslcmop_operation_state
= "FAILED"
5160 self
._write
_ns
_status
(
5164 ], # TODO check if degraded. For the moment use previous status
5165 current_operation
="IDLE",
5166 current_operation_id
=None,
5167 # error_description=error_description_nsr,
5168 # error_detail=error_detail,
5169 other_update
=db_nsr_update
,
5172 self
._write
_op
_status
(
5175 error_message
=error_description_nslcmop
,
5176 operation_state
=nslcmop_operation_state
,
5177 other_update
=db_nslcmop_update
,
5180 if nslcmop_operation_state
:
5182 await self
.msg
.aiowrite(
5187 "nslcmop_id": nslcmop_id
,
5188 "operationState": nslcmop_operation_state
,
5192 except Exception as e
:
5194 logging_text
+ "kafka_write notification Exception {}".format(e
)
5196 self
.logger
.debug(logging_text
+ "Exit")
5197 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5198 return nslcmop_operation_state
, detailed_status
5200 async def scale(self
, nsr_id
, nslcmop_id
):
5201 # Try to lock HA task here
5202 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5203 if not task_is_locked_by_me
:
5206 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5207 stage
= ["", "", ""]
5208 tasks_dict_info
= {}
5209 # ^ stage, step, VIM progress
5210 self
.logger
.debug(logging_text
+ "Enter")
5211 # get all needed from database
5213 db_nslcmop_update
= {}
5216 # in case of error, indicates what part of scale was failed to put nsr at error status
5217 scale_process
= None
5218 old_operational_status
= ""
5219 old_config_status
= ""
5222 # wait for any previous tasks in process
5223 step
= "Waiting for previous operations to terminate"
5224 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5225 self
._write
_ns
_status
(
5228 current_operation
="SCALING",
5229 current_operation_id
=nslcmop_id
,
5232 step
= "Getting nslcmop from database"
5234 step
+ " after having waited for previous tasks to be completed"
5236 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5238 step
= "Getting nsr from database"
5239 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5240 old_operational_status
= db_nsr
["operational-status"]
5241 old_config_status
= db_nsr
["config-status"]
5243 step
= "Parsing scaling parameters"
5244 db_nsr_update
["operational-status"] = "scaling"
5245 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5246 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5248 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5250 ]["member-vnf-index"]
5251 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5253 ]["scaling-group-descriptor"]
5254 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5255 # for backward compatibility
5256 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5257 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5258 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5259 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5261 step
= "Getting vnfr from database"
5262 db_vnfr
= self
.db
.get_one(
5263 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5266 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5268 step
= "Getting vnfd from database"
5269 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5271 base_folder
= db_vnfd
["_admin"]["storage"]
5273 step
= "Getting scaling-group-descriptor"
5274 scaling_descriptor
= find_in_list(
5275 get_scaling_aspect(db_vnfd
),
5276 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5278 if not scaling_descriptor
:
5280 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5281 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5284 step
= "Sending scale order to VIM"
5285 # TODO check if ns is in a proper status
5287 if not db_nsr
["_admin"].get("scaling-group"):
5292 "_admin.scaling-group": [
5293 {"name": scaling_group
, "nb-scale-op": 0}
5297 admin_scale_index
= 0
5299 for admin_scale_index
, admin_scale_info
in enumerate(
5300 db_nsr
["_admin"]["scaling-group"]
5302 if admin_scale_info
["name"] == scaling_group
:
5303 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5305 else: # not found, set index one plus last element and add new entry with the name
5306 admin_scale_index
+= 1
5308 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5311 vca_scaling_info
= []
5312 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5313 if scaling_type
== "SCALE_OUT":
5314 if "aspect-delta-details" not in scaling_descriptor
:
5316 "Aspect delta details not fount in scaling descriptor {}".format(
5317 scaling_descriptor
["name"]
5320 # count if max-instance-count is reached
5321 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5323 scaling_info
["scaling_direction"] = "OUT"
5324 scaling_info
["vdu-create"] = {}
5325 scaling_info
["kdu-create"] = {}
5326 for delta
in deltas
:
5327 for vdu_delta
in delta
.get("vdu-delta", {}):
5328 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5329 # vdu_index also provides the number of instance of the targeted vdu
5330 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5331 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5335 additional_params
= (
5336 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5339 cloud_init_list
= []
5341 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5342 max_instance_count
= 10
5343 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5344 max_instance_count
= vdu_profile
.get(
5345 "max-number-of-instances", 10
5348 default_instance_num
= get_number_of_instances(
5351 instances_number
= vdu_delta
.get("number-of-instances", 1)
5352 nb_scale_op
+= instances_number
5354 new_instance_count
= nb_scale_op
+ default_instance_num
5355 # Control if new count is over max and vdu count is less than max.
5356 # Then assign new instance count
5357 if new_instance_count
> max_instance_count
> vdu_count
:
5358 instances_number
= new_instance_count
- max_instance_count
5360 instances_number
= instances_number
5362 if new_instance_count
> max_instance_count
:
5364 "reached the limit of {} (max-instance-count) "
5365 "scaling-out operations for the "
5366 "scaling-group-descriptor '{}'".format(
5367 nb_scale_op
, scaling_group
5370 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5372 # TODO Information of its own ip is not available because db_vnfr is not updated.
5373 additional_params
["OSM"] = get_osm_params(
5374 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5376 cloud_init_list
.append(
5377 self
._parse
_cloud
_init
(
5384 vca_scaling_info
.append(
5386 "osm_vdu_id": vdu_delta
["id"],
5387 "member-vnf-index": vnf_index
,
5389 "vdu_index": vdu_index
+ x
,
5392 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5393 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5394 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5395 kdu_name
= kdu_profile
["kdu-name"]
5396 resource_name
= kdu_profile
.get("resource-name", "")
5398 # Might have different kdus in the same delta
5399 # Should have list for each kdu
5400 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5401 scaling_info
["kdu-create"][kdu_name
] = []
5403 kdur
= get_kdur(db_vnfr
, kdu_name
)
5404 if kdur
.get("helm-chart"):
5405 k8s_cluster_type
= "helm-chart-v3"
5406 self
.logger
.debug("kdur: {}".format(kdur
))
5408 kdur
.get("helm-version")
5409 and kdur
.get("helm-version") == "v2"
5411 k8s_cluster_type
= "helm-chart"
5412 elif kdur
.get("juju-bundle"):
5413 k8s_cluster_type
= "juju-bundle"
5416 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5417 "juju-bundle. Maybe an old NBI version is running".format(
5418 db_vnfr
["member-vnf-index-ref"], kdu_name
5422 max_instance_count
= 10
5423 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5424 max_instance_count
= kdu_profile
.get(
5425 "max-number-of-instances", 10
5428 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5429 deployed_kdu
, _
= get_deployed_kdu(
5430 nsr_deployed
, kdu_name
, vnf_index
5432 if deployed_kdu
is None:
5434 "KDU '{}' for vnf '{}' not deployed".format(
5438 kdu_instance
= deployed_kdu
.get("kdu-instance")
5439 instance_num
= await self
.k8scluster_map
[
5445 cluster_uuid
=deployed_kdu
.get("k8scluster-uuid"),
5446 kdu_model
=deployed_kdu
.get("kdu-model"),
5448 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5449 "number-of-instances", 1
5452 # Control if new count is over max and instance_num is less than max.
5453 # Then assign max instance number to kdu replica count
5454 if kdu_replica_count
> max_instance_count
> instance_num
:
5455 kdu_replica_count
= max_instance_count
5456 if kdu_replica_count
> max_instance_count
:
5458 "reached the limit of {} (max-instance-count) "
5459 "scaling-out operations for the "
5460 "scaling-group-descriptor '{}'".format(
5461 instance_num
, scaling_group
5465 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5466 vca_scaling_info
.append(
5468 "osm_kdu_id": kdu_name
,
5469 "member-vnf-index": vnf_index
,
5471 "kdu_index": instance_num
+ x
- 1,
5474 scaling_info
["kdu-create"][kdu_name
].append(
5476 "member-vnf-index": vnf_index
,
5478 "k8s-cluster-type": k8s_cluster_type
,
5479 "resource-name": resource_name
,
5480 "scale": kdu_replica_count
,
5483 elif scaling_type
== "SCALE_IN":
5484 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5486 scaling_info
["scaling_direction"] = "IN"
5487 scaling_info
["vdu-delete"] = {}
5488 scaling_info
["kdu-delete"] = {}
5490 for delta
in deltas
:
5491 for vdu_delta
in delta
.get("vdu-delta", {}):
5492 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5493 min_instance_count
= 0
5494 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5495 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5496 min_instance_count
= vdu_profile
["min-number-of-instances"]
5498 default_instance_num
= get_number_of_instances(
5499 db_vnfd
, vdu_delta
["id"]
5501 instance_num
= vdu_delta
.get("number-of-instances", 1)
5502 nb_scale_op
-= instance_num
5504 new_instance_count
= nb_scale_op
+ default_instance_num
5506 if new_instance_count
< min_instance_count
< vdu_count
:
5507 instances_number
= min_instance_count
- new_instance_count
5509 instances_number
= instance_num
5511 if new_instance_count
< min_instance_count
:
5513 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5514 "scaling-group-descriptor '{}'".format(
5515 nb_scale_op
, scaling_group
5518 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5519 vca_scaling_info
.append(
5521 "osm_vdu_id": vdu_delta
["id"],
5522 "member-vnf-index": vnf_index
,
5524 "vdu_index": vdu_index
- 1 - x
,
5527 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5528 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5529 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5530 kdu_name
= kdu_profile
["kdu-name"]
5531 resource_name
= kdu_profile
.get("resource-name", "")
5533 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5534 scaling_info
["kdu-delete"][kdu_name
] = []
5536 kdur
= get_kdur(db_vnfr
, kdu_name
)
5537 if kdur
.get("helm-chart"):
5538 k8s_cluster_type
= "helm-chart-v3"
5539 self
.logger
.debug("kdur: {}".format(kdur
))
5541 kdur
.get("helm-version")
5542 and kdur
.get("helm-version") == "v2"
5544 k8s_cluster_type
= "helm-chart"
5545 elif kdur
.get("juju-bundle"):
5546 k8s_cluster_type
= "juju-bundle"
5549 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5550 "juju-bundle. Maybe an old NBI version is running".format(
5551 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5555 min_instance_count
= 0
5556 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5557 min_instance_count
= kdu_profile
["min-number-of-instances"]
5559 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5560 deployed_kdu
, _
= get_deployed_kdu(
5561 nsr_deployed
, kdu_name
, vnf_index
5563 if deployed_kdu
is None:
5565 "KDU '{}' for vnf '{}' not deployed".format(
5569 kdu_instance
= deployed_kdu
.get("kdu-instance")
5570 instance_num
= await self
.k8scluster_map
[
5576 cluster_uuid
=deployed_kdu
.get("k8scluster-uuid"),
5577 kdu_model
=deployed_kdu
.get("kdu-model"),
5579 kdu_replica_count
= instance_num
- kdu_delta
.get(
5580 "number-of-instances", 1
5583 if kdu_replica_count
< min_instance_count
< instance_num
:
5584 kdu_replica_count
= min_instance_count
5585 if kdu_replica_count
< min_instance_count
:
5587 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5588 "scaling-group-descriptor '{}'".format(
5589 instance_num
, scaling_group
5593 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5594 vca_scaling_info
.append(
5596 "osm_kdu_id": kdu_name
,
5597 "member-vnf-index": vnf_index
,
5599 "kdu_index": instance_num
- x
- 1,
5602 scaling_info
["kdu-delete"][kdu_name
].append(
5604 "member-vnf-index": vnf_index
,
5606 "k8s-cluster-type": k8s_cluster_type
,
5607 "resource-name": resource_name
,
5608 "scale": kdu_replica_count
,
5612 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5613 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5614 if scaling_info
["scaling_direction"] == "IN":
5615 for vdur
in reversed(db_vnfr
["vdur"]):
5616 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5617 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5618 scaling_info
["vdu"].append(
5620 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5621 "vdu_id": vdur
["vdu-id-ref"],
5625 for interface
in vdur
["interfaces"]:
5626 scaling_info
["vdu"][-1]["interface"].append(
5628 "name": interface
["name"],
5629 "ip_address": interface
["ip-address"],
5630 "mac_address": interface
.get("mac-address"),
5633 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5636 step
= "Executing pre-scale vnf-config-primitive"
5637 if scaling_descriptor
.get("scaling-config-action"):
5638 for scaling_config_action
in scaling_descriptor
[
5639 "scaling-config-action"
5642 scaling_config_action
.get("trigger") == "pre-scale-in"
5643 and scaling_type
== "SCALE_IN"
5645 scaling_config_action
.get("trigger") == "pre-scale-out"
5646 and scaling_type
== "SCALE_OUT"
5648 vnf_config_primitive
= scaling_config_action
[
5649 "vnf-config-primitive-name-ref"
5651 step
= db_nslcmop_update
[
5653 ] = "executing pre-scale scaling-config-action '{}'".format(
5654 vnf_config_primitive
5657 # look for primitive
5658 for config_primitive
in (
5659 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5660 ).get("config-primitive", ()):
5661 if config_primitive
["name"] == vnf_config_primitive
:
5665 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5666 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5667 "primitive".format(scaling_group
, vnf_config_primitive
)
5670 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5671 if db_vnfr
.get("additionalParamsForVnf"):
5672 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5674 scale_process
= "VCA"
5675 db_nsr_update
["config-status"] = "configuring pre-scaling"
5676 primitive_params
= self
._map
_primitive
_params
(
5677 config_primitive
, {}, vnfr_params
5680 # Pre-scale retry check: Check if this sub-operation has been executed before
5681 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5684 vnf_config_primitive
,
5688 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5689 # Skip sub-operation
5690 result
= "COMPLETED"
5691 result_detail
= "Done"
5694 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5695 vnf_config_primitive
, result
, result_detail
5699 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5700 # New sub-operation: Get index of this sub-operation
5702 len(db_nslcmop
.get("_admin", {}).get("operations"))
5707 + "vnf_config_primitive={} New sub-operation".format(
5708 vnf_config_primitive
5712 # retry: Get registered params for this existing sub-operation
5713 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5716 vnf_index
= op
.get("member_vnf_index")
5717 vnf_config_primitive
= op
.get("primitive")
5718 primitive_params
= op
.get("primitive_params")
5721 + "vnf_config_primitive={} Sub-operation retry".format(
5722 vnf_config_primitive
5725 # Execute the primitive, either with new (first-time) or registered (reintent) args
5726 ee_descriptor_id
= config_primitive
.get(
5727 "execution-environment-ref"
5729 primitive_name
= config_primitive
.get(
5730 "execution-environment-primitive", vnf_config_primitive
5732 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5733 nsr_deployed
["VCA"],
5734 member_vnf_index
=vnf_index
,
5736 vdu_count_index
=None,
5737 ee_descriptor_id
=ee_descriptor_id
,
5739 result
, result_detail
= await self
._ns
_execute
_primitive
(
5748 + "vnf_config_primitive={} Done with result {} {}".format(
5749 vnf_config_primitive
, result
, result_detail
5752 # Update operationState = COMPLETED | FAILED
5753 self
._update
_suboperation
_status
(
5754 db_nslcmop
, op_index
, result
, result_detail
5757 if result
== "FAILED":
5758 raise LcmException(result_detail
)
5759 db_nsr_update
["config-status"] = old_config_status
5760 scale_process
= None
5764 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5767 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5770 # SCALE-IN VCA - BEGIN
5771 if vca_scaling_info
:
5772 step
= db_nslcmop_update
[
5774 ] = "Deleting the execution environments"
5775 scale_process
= "VCA"
5776 for vca_info
in vca_scaling_info
:
5777 if vca_info
["type"] == "delete" and not vca_info
.get("osm_kdu_id"):
5778 member_vnf_index
= str(vca_info
["member-vnf-index"])
5780 logging_text
+ "vdu info: {}".format(vca_info
)
5782 if vca_info
.get("osm_vdu_id"):
5783 vdu_id
= vca_info
["osm_vdu_id"]
5784 vdu_index
= int(vca_info
["vdu_index"])
5787 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5788 member_vnf_index
, vdu_id
, vdu_index
5790 stage
[2] = step
= "Scaling in VCA"
5791 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5792 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5793 config_update
= db_nsr
["configurationStatus"]
5794 for vca_index
, vca
in enumerate(vca_update
):
5796 (vca
or vca
.get("ee_id"))
5797 and vca
["member-vnf-index"] == member_vnf_index
5798 and vca
["vdu_count_index"] == vdu_index
5800 if vca
.get("vdu_id"):
5801 config_descriptor
= get_configuration(
5802 db_vnfd
, vca
.get("vdu_id")
5804 elif vca
.get("kdu_name"):
5805 config_descriptor
= get_configuration(
5806 db_vnfd
, vca
.get("kdu_name")
5809 config_descriptor
= get_configuration(
5810 db_vnfd
, db_vnfd
["id"]
5812 operation_params
= (
5813 db_nslcmop
.get("operationParams") or {}
5815 exec_terminate_primitives
= not operation_params
.get(
5816 "skip_terminate_primitives"
5817 ) and vca
.get("needed_terminate")
5818 task
= asyncio
.ensure_future(
5827 exec_primitives
=exec_terminate_primitives
,
5831 timeout
=self
.timeout_charm_delete
,
5834 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5837 del vca_update
[vca_index
]
5838 del config_update
[vca_index
]
5839 # wait for pending tasks of terminate primitives
5843 + "Waiting for tasks {}".format(
5844 list(tasks_dict_info
.keys())
5847 error_list
= await self
._wait
_for
_tasks
(
5851 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5856 tasks_dict_info
.clear()
5858 raise LcmException("; ".join(error_list
))
5860 db_vca_and_config_update
= {
5861 "_admin.deployed.VCA": vca_update
,
5862 "configurationStatus": config_update
,
5865 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5867 scale_process
= None
5868 # SCALE-IN VCA - END
5871 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5872 scale_process
= "RO"
5873 if self
.ro_config
.get("ng"):
5874 await self
._scale
_ng
_ro
(
5875 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5877 scaling_info
.pop("vdu-create", None)
5878 scaling_info
.pop("vdu-delete", None)
5880 scale_process
= None
5884 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5885 scale_process
= "KDU"
5886 await self
._scale
_kdu
(
5887 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5889 scaling_info
.pop("kdu-create", None)
5890 scaling_info
.pop("kdu-delete", None)
5892 scale_process
= None
5896 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5898 # SCALE-UP VCA - BEGIN
5899 if vca_scaling_info
:
5900 step
= db_nslcmop_update
[
5902 ] = "Creating new execution environments"
5903 scale_process
= "VCA"
5904 for vca_info
in vca_scaling_info
:
5905 if vca_info
["type"] == "create" and not vca_info
.get("osm_kdu_id"):
5906 member_vnf_index
= str(vca_info
["member-vnf-index"])
5908 logging_text
+ "vdu info: {}".format(vca_info
)
5910 vnfd_id
= db_vnfr
["vnfd-ref"]
5911 if vca_info
.get("osm_vdu_id"):
5912 vdu_index
= int(vca_info
["vdu_index"])
5913 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5914 if db_vnfr
.get("additionalParamsForVnf"):
5915 deploy_params
.update(
5917 db_vnfr
["additionalParamsForVnf"].copy()
5920 descriptor_config
= get_configuration(
5921 db_vnfd
, db_vnfd
["id"]
5923 if descriptor_config
:
5928 logging_text
=logging_text
5929 + "member_vnf_index={} ".format(member_vnf_index
),
5932 nslcmop_id
=nslcmop_id
,
5938 member_vnf_index
=member_vnf_index
,
5939 vdu_index
=vdu_index
,
5941 deploy_params
=deploy_params
,
5942 descriptor_config
=descriptor_config
,
5943 base_folder
=base_folder
,
5944 task_instantiation_info
=tasks_dict_info
,
5947 vdu_id
= vca_info
["osm_vdu_id"]
5948 vdur
= find_in_list(
5949 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5951 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5952 if vdur
.get("additionalParams"):
5953 deploy_params_vdu
= parse_yaml_strings(
5954 vdur
["additionalParams"]
5957 deploy_params_vdu
= deploy_params
5958 deploy_params_vdu
["OSM"] = get_osm_params(
5959 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5961 if descriptor_config
:
5966 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5967 member_vnf_index
, vdu_id
, vdu_index
5969 stage
[2] = step
= "Scaling out VCA"
5970 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5972 logging_text
=logging_text
5973 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5974 member_vnf_index
, vdu_id
, vdu_index
5978 nslcmop_id
=nslcmop_id
,
5984 member_vnf_index
=member_vnf_index
,
5985 vdu_index
=vdu_index
,
5987 deploy_params
=deploy_params_vdu
,
5988 descriptor_config
=descriptor_config
,
5989 base_folder
=base_folder
,
5990 task_instantiation_info
=tasks_dict_info
,
5993 # SCALE-UP VCA - END
5994 scale_process
= None
5997 # execute primitive service POST-SCALING
5998 step
= "Executing post-scale vnf-config-primitive"
5999 if scaling_descriptor
.get("scaling-config-action"):
6000 for scaling_config_action
in scaling_descriptor
[
6001 "scaling-config-action"
6004 scaling_config_action
.get("trigger") == "post-scale-in"
6005 and scaling_type
== "SCALE_IN"
6007 scaling_config_action
.get("trigger") == "post-scale-out"
6008 and scaling_type
== "SCALE_OUT"
6010 vnf_config_primitive
= scaling_config_action
[
6011 "vnf-config-primitive-name-ref"
6013 step
= db_nslcmop_update
[
6015 ] = "executing post-scale scaling-config-action '{}'".format(
6016 vnf_config_primitive
6019 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
6020 if db_vnfr
.get("additionalParamsForVnf"):
6021 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
6023 # look for primitive
6024 for config_primitive
in (
6025 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
6026 ).get("config-primitive", ()):
6027 if config_primitive
["name"] == vnf_config_primitive
:
6031 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
6032 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
6033 "config-primitive".format(
6034 scaling_group
, vnf_config_primitive
6037 scale_process
= "VCA"
6038 db_nsr_update
["config-status"] = "configuring post-scaling"
6039 primitive_params
= self
._map
_primitive
_params
(
6040 config_primitive
, {}, vnfr_params
6043 # Post-scale retry check: Check if this sub-operation has been executed before
6044 op_index
= self
._check
_or
_add
_scale
_suboperation
(
6047 vnf_config_primitive
,
6051 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
6052 # Skip sub-operation
6053 result
= "COMPLETED"
6054 result_detail
= "Done"
6057 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
6058 vnf_config_primitive
, result
, result_detail
6062 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
6063 # New sub-operation: Get index of this sub-operation
6065 len(db_nslcmop
.get("_admin", {}).get("operations"))
6070 + "vnf_config_primitive={} New sub-operation".format(
6071 vnf_config_primitive
6075 # retry: Get registered params for this existing sub-operation
6076 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
6079 vnf_index
= op
.get("member_vnf_index")
6080 vnf_config_primitive
= op
.get("primitive")
6081 primitive_params
= op
.get("primitive_params")
6084 + "vnf_config_primitive={} Sub-operation retry".format(
6085 vnf_config_primitive
6088 # Execute the primitive, either with new (first-time) or registered (reintent) args
6089 ee_descriptor_id
= config_primitive
.get(
6090 "execution-environment-ref"
6092 primitive_name
= config_primitive
.get(
6093 "execution-environment-primitive", vnf_config_primitive
6095 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6096 nsr_deployed
["VCA"],
6097 member_vnf_index
=vnf_index
,
6099 vdu_count_index
=None,
6100 ee_descriptor_id
=ee_descriptor_id
,
6102 result
, result_detail
= await self
._ns
_execute
_primitive
(
6111 + "vnf_config_primitive={} Done with result {} {}".format(
6112 vnf_config_primitive
, result
, result_detail
6115 # Update operationState = COMPLETED | FAILED
6116 self
._update
_suboperation
_status
(
6117 db_nslcmop
, op_index
, result
, result_detail
6120 if result
== "FAILED":
6121 raise LcmException(result_detail
)
6122 db_nsr_update
["config-status"] = old_config_status
6123 scale_process
= None
6128 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6129 db_nsr_update
["operational-status"] = (
6131 if old_operational_status
== "failed"
6132 else old_operational_status
6134 db_nsr_update
["config-status"] = old_config_status
6137 ROclient
.ROClientException
,
6142 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6144 except asyncio
.CancelledError
:
6146 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6148 exc
= "Operation was cancelled"
6149 except Exception as e
:
6150 exc
= traceback
.format_exc()
6151 self
.logger
.critical(
6152 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6156 self
._write
_ns
_status
(
6159 current_operation
="IDLE",
6160 current_operation_id
=None,
6163 stage
[1] = "Waiting for instantiate pending tasks."
6164 self
.logger
.debug(logging_text
+ stage
[1])
6165 exc
= await self
._wait
_for
_tasks
(
6168 self
.timeout_ns_deploy
,
6176 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6177 nslcmop_operation_state
= "FAILED"
6179 db_nsr_update
["operational-status"] = old_operational_status
6180 db_nsr_update
["config-status"] = old_config_status
6181 db_nsr_update
["detailed-status"] = ""
6183 if "VCA" in scale_process
:
6184 db_nsr_update
["config-status"] = "failed"
6185 if "RO" in scale_process
:
6186 db_nsr_update
["operational-status"] = "failed"
6189 ] = "FAILED scaling nslcmop={} {}: {}".format(
6190 nslcmop_id
, step
, exc
6193 error_description_nslcmop
= None
6194 nslcmop_operation_state
= "COMPLETED"
6195 db_nslcmop_update
["detailed-status"] = "Done"
6197 self
._write
_op
_status
(
6200 error_message
=error_description_nslcmop
,
6201 operation_state
=nslcmop_operation_state
,
6202 other_update
=db_nslcmop_update
,
6205 self
._write
_ns
_status
(
6208 current_operation
="IDLE",
6209 current_operation_id
=None,
6210 other_update
=db_nsr_update
,
6213 if nslcmop_operation_state
:
6217 "nslcmop_id": nslcmop_id
,
6218 "operationState": nslcmop_operation_state
,
6220 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6221 except Exception as e
:
6223 logging_text
+ "kafka_write notification Exception {}".format(e
)
6225 self
.logger
.debug(logging_text
+ "Exit")
6226 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6228 async def _scale_kdu(
6229 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6231 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6232 for kdu_name
in _scaling_info
:
6233 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6234 deployed_kdu
, index
= get_deployed_kdu(
6235 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6237 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6238 kdu_instance
= deployed_kdu
["kdu-instance"]
6239 kdu_model
= deployed_kdu
.get("kdu-model")
6240 scale
= int(kdu_scaling_info
["scale"])
6241 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6244 "collection": "nsrs",
6245 "filter": {"_id": nsr_id
},
6246 "path": "_admin.deployed.K8s.{}".format(index
),
6249 step
= "scaling application {}".format(
6250 kdu_scaling_info
["resource-name"]
6252 self
.logger
.debug(logging_text
+ step
)
6254 if kdu_scaling_info
["type"] == "delete":
6255 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6258 and kdu_config
.get("terminate-config-primitive")
6259 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6261 terminate_config_primitive_list
= kdu_config
.get(
6262 "terminate-config-primitive"
6264 terminate_config_primitive_list
.sort(
6265 key
=lambda val
: int(val
["seq"])
6269 terminate_config_primitive
6270 ) in terminate_config_primitive_list
:
6271 primitive_params_
= self
._map
_primitive
_params
(
6272 terminate_config_primitive
, {}, {}
6274 step
= "execute terminate config primitive"
6275 self
.logger
.debug(logging_text
+ step
)
6276 await asyncio
.wait_for(
6277 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6278 cluster_uuid
=cluster_uuid
,
6279 kdu_instance
=kdu_instance
,
6280 primitive_name
=terminate_config_primitive
["name"],
6281 params
=primitive_params_
,
6288 await asyncio
.wait_for(
6289 self
.k8scluster_map
[k8s_cluster_type
].scale(
6292 kdu_scaling_info
["resource-name"],
6294 cluster_uuid
=cluster_uuid
,
6295 kdu_model
=kdu_model
,
6299 timeout
=self
.timeout_vca_on_error
,
6302 if kdu_scaling_info
["type"] == "create":
6303 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6306 and kdu_config
.get("initial-config-primitive")
6307 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6309 initial_config_primitive_list
= kdu_config
.get(
6310 "initial-config-primitive"
6312 initial_config_primitive_list
.sort(
6313 key
=lambda val
: int(val
["seq"])
6316 for initial_config_primitive
in initial_config_primitive_list
:
6317 primitive_params_
= self
._map
_primitive
_params
(
6318 initial_config_primitive
, {}, {}
6320 step
= "execute initial config primitive"
6321 self
.logger
.debug(logging_text
+ step
)
6322 await asyncio
.wait_for(
6323 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6324 cluster_uuid
=cluster_uuid
,
6325 kdu_instance
=kdu_instance
,
6326 primitive_name
=initial_config_primitive
["name"],
6327 params
=primitive_params_
,
6334 async def _scale_ng_ro(
6335 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6337 nsr_id
= db_nslcmop
["nsInstanceId"]
6338 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6341 # read from db: vnfd's for every vnf
6344 # for each vnf in ns, read vnfd
6345 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6346 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6347 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6348 # if we haven't this vnfd, read it from db
6349 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6351 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6352 db_vnfds
.append(vnfd
)
6353 n2vc_key
= self
.n2vc
.get_public_key()
6354 n2vc_key_list
= [n2vc_key
]
6357 vdu_scaling_info
.get("vdu-create"),
6358 vdu_scaling_info
.get("vdu-delete"),
6361 # db_vnfr has been updated, update db_vnfrs to use it
6362 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6363 await self
._instantiate
_ng
_ro
(
6373 start_deploy
=time(),
6374 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6376 if vdu_scaling_info
.get("vdu-delete"):
6378 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6381 async def extract_prometheus_scrape_jobs(
6382 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6384 # look if exist a file called 'prometheus*.j2' and
6385 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6389 for f
in artifact_content
6390 if f
.startswith("prometheus") and f
.endswith(".j2")
6396 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6400 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6401 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6403 vnfr_id
= vnfr_id
.replace("-", "")
6405 "JOB_NAME": vnfr_id
,
6406 "TARGET_IP": target_ip
,
6407 "EXPORTER_POD_IP": host_name
,
6408 "EXPORTER_POD_PORT": host_port
,
6410 job_list
= parse_job(job_data
, variables
)
6411 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6412 for job
in job_list
:
6414 not isinstance(job
.get("job_name"), str)
6415 or vnfr_id
not in job
["job_name"]
6417 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6418 job
["nsr_id"] = nsr_id
6419 job
["vnfr_id"] = vnfr_id
6422 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6424 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6426 :param: vim_account_id: VIM Account ID
6428 :return: (cloud_name, cloud_credential)
6430 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6431 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6433 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6435 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6437 :param: vim_account_id: VIM Account ID
6439 :return: (cloud_name, cloud_credential)
6441 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6442 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")