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 nslcmop_id
= db_nslcmop
["_id"]
827 "name": db_nsr
["name"],
830 "image": deepcopy(db_nsr
["image"]),
831 "flavor": deepcopy(db_nsr
["flavor"]),
832 "action_id": nslcmop_id
,
833 "cloud_init_content": {},
835 for image
in target
["image"]:
836 image
["vim_info"] = {}
837 for flavor
in target
["flavor"]:
838 flavor
["vim_info"] = {}
840 if db_nslcmop
.get("lcmOperationType") != "instantiate":
841 # get parameters of instantiation:
842 db_nslcmop_instantiate
= self
.db
.get_list(
845 "nsInstanceId": db_nslcmop
["nsInstanceId"],
846 "lcmOperationType": "instantiate",
849 ns_params
= db_nslcmop_instantiate
.get("operationParams")
851 ns_params
= db_nslcmop
.get("operationParams")
852 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
853 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
856 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
857 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
861 "mgmt-network": vld
.get("mgmt-network", False),
862 "type": vld
.get("type"),
865 "vim_network_name": vld
.get("vim-network-name"),
866 "vim_account_id": ns_params
["vimAccountId"],
870 # check if this network needs SDN assist
871 if vld
.get("pci-interfaces"):
872 db_vim
= get_vim_account(ns_params
["vimAccountId"])
873 sdnc_id
= db_vim
["config"].get("sdn-controller")
875 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
876 target_sdn
= "sdn:{}".format(sdnc_id
)
877 target_vld
["vim_info"][target_sdn
] = {
879 "target_vim": target_vim
,
881 "type": vld
.get("type"),
884 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
885 for nsd_vnf_profile
in nsd_vnf_profiles
:
886 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
887 if cp
["virtual-link-profile-id"] == vld
["id"]:
889 "member_vnf:{}.{}".format(
890 cp
["constituent-cpd-id"][0][
891 "constituent-base-element-id"
893 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
895 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
897 # check at nsd descriptor, if there is an ip-profile
899 nsd_vlp
= find_in_list(
900 get_virtual_link_profiles(nsd
),
901 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
906 and nsd_vlp
.get("virtual-link-protocol-data")
907 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
909 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
912 ip_profile_dest_data
= {}
913 if "ip-version" in ip_profile_source_data
:
914 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
917 if "cidr" in ip_profile_source_data
:
918 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
921 if "gateway-ip" in ip_profile_source_data
:
922 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
925 if "dhcp-enabled" in ip_profile_source_data
:
926 ip_profile_dest_data
["dhcp-params"] = {
927 "enabled": ip_profile_source_data
["dhcp-enabled"]
929 vld_params
["ip-profile"] = ip_profile_dest_data
931 # update vld_params with instantiation params
932 vld_instantiation_params
= find_in_list(
933 get_iterable(ns_params
, "vld"),
934 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
936 if vld_instantiation_params
:
937 vld_params
.update(vld_instantiation_params
)
938 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
939 target
["ns"]["vld"].append(target_vld
)
941 for vnfr
in db_vnfrs
.values():
943 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
945 vnf_params
= find_in_list(
946 get_iterable(ns_params
, "vnf"),
947 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
949 target_vnf
= deepcopy(vnfr
)
950 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
951 for vld
in target_vnf
.get("vld", ()):
952 # check if connected to a ns.vld, to fill target'
953 vnf_cp
= find_in_list(
954 vnfd
.get("int-virtual-link-desc", ()),
955 lambda cpd
: cpd
.get("id") == vld
["id"],
958 ns_cp
= "member_vnf:{}.{}".format(
959 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
961 if cp2target
.get(ns_cp
):
962 vld
["target"] = cp2target
[ns_cp
]
965 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
967 # check if this network needs SDN assist
969 if vld
.get("pci-interfaces"):
970 db_vim
= get_vim_account(vnfr
["vim-account-id"])
971 sdnc_id
= db_vim
["config"].get("sdn-controller")
973 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
974 target_sdn
= "sdn:{}".format(sdnc_id
)
975 vld
["vim_info"][target_sdn
] = {
977 "target_vim": target_vim
,
979 "type": vld
.get("type"),
982 # check at vnfd descriptor, if there is an ip-profile
984 vnfd_vlp
= find_in_list(
985 get_virtual_link_profiles(vnfd
),
986 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
990 and vnfd_vlp
.get("virtual-link-protocol-data")
991 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
993 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
996 ip_profile_dest_data
= {}
997 if "ip-version" in ip_profile_source_data
:
998 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
1001 if "cidr" in ip_profile_source_data
:
1002 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
1005 if "gateway-ip" in ip_profile_source_data
:
1006 ip_profile_dest_data
[
1008 ] = ip_profile_source_data
["gateway-ip"]
1009 if "dhcp-enabled" in ip_profile_source_data
:
1010 ip_profile_dest_data
["dhcp-params"] = {
1011 "enabled": ip_profile_source_data
["dhcp-enabled"]
1014 vld_params
["ip-profile"] = ip_profile_dest_data
1015 # update vld_params with instantiation params
1017 vld_instantiation_params
= find_in_list(
1018 get_iterable(vnf_params
, "internal-vld"),
1019 lambda i_vld
: i_vld
["name"] == vld
["id"],
1021 if vld_instantiation_params
:
1022 vld_params
.update(vld_instantiation_params
)
1023 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1026 for vdur
in target_vnf
.get("vdur", ()):
1027 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1028 continue # This vdu must not be created
1029 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1031 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1034 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1035 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1038 and vdu_configuration
.get("config-access")
1039 and vdu_configuration
.get("config-access").get("ssh-access")
1041 vdur
["ssh-keys"] = ssh_keys_all
1042 vdur
["ssh-access-required"] = vdu_configuration
[
1044 ]["ssh-access"]["required"]
1047 and vnf_configuration
.get("config-access")
1048 and vnf_configuration
.get("config-access").get("ssh-access")
1049 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1051 vdur
["ssh-keys"] = ssh_keys_all
1052 vdur
["ssh-access-required"] = vnf_configuration
[
1054 ]["ssh-access"]["required"]
1055 elif ssh_keys_instantiation
and find_in_list(
1056 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1058 vdur
["ssh-keys"] = ssh_keys_instantiation
1060 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1062 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1064 if vdud
.get("cloud-init-file"):
1065 vdur
["cloud-init"] = "{}:file:{}".format(
1066 vnfd
["_id"], vdud
.get("cloud-init-file")
1068 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1069 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1070 base_folder
= vnfd
["_admin"]["storage"]
1071 if base_folder
["pkg-dir"]:
1072 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1073 base_folder
["folder"],
1074 base_folder
["pkg-dir"],
1075 vdud
.get("cloud-init-file"),
1078 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
1079 base_folder
["folder"],
1080 vdud
.get("cloud-init-file"),
1082 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
1083 target
["cloud_init_content"][
1086 elif vdud
.get("cloud-init"):
1087 vdur
["cloud-init"] = "{}:vdu:{}".format(
1088 vnfd
["_id"], get_vdu_index(vnfd
, vdur
["vdu-id-ref"])
1090 # put content at target.cloul_init_content. Avoid ng_ro read vnfd descriptor
1091 target
["cloud_init_content"][vdur
["cloud-init"]] = vdud
[
1094 vdur
["additionalParams"] = vdur
.get("additionalParams") or {}
1095 deploy_params_vdu
= self
._format
_additional
_params
(
1096 vdur
.get("additionalParams") or {}
1098 deploy_params_vdu
["OSM"] = get_osm_params(
1099 vnfr
, vdur
["vdu-id-ref"], vdur
["count-index"]
1101 vdur
["additionalParams"] = deploy_params_vdu
1104 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
1105 if target_vim
not in ns_flavor
["vim_info"]:
1106 ns_flavor
["vim_info"][target_vim
] = {}
1109 # in case alternative images are provided we must check if they should be applied
1110 # for the vim_type, modify the vim_type taking into account
1111 ns_image_id
= int(vdur
["ns-image-id"])
1112 if vdur
.get("alt-image-ids"):
1113 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1114 vim_type
= db_vim
["vim_type"]
1115 for alt_image_id
in vdur
.get("alt-image-ids"):
1116 ns_alt_image
= target
["image"][int(alt_image_id
)]
1117 if vim_type
== ns_alt_image
.get("vim-type"):
1118 # must use alternative image
1120 "use alternative image id: {}".format(alt_image_id
)
1122 ns_image_id
= alt_image_id
1123 vdur
["ns-image-id"] = ns_image_id
1125 ns_image
= target
["image"][int(ns_image_id
)]
1126 if target_vim
not in ns_image
["vim_info"]:
1127 ns_image
["vim_info"][target_vim
] = {}
1129 vdur
["vim_info"] = {target_vim
: {}}
1130 # instantiation parameters
1132 # vdu_instantiation_params = next((v for v in get_iterable(vnf_params, "vdu") if v["id"] ==
1133 # vdud["id"]), None)
1134 vdur_list
.append(vdur
)
1135 target_vnf
["vdur"] = vdur_list
1136 target
["vnf"].append(target_vnf
)
1138 desc
= await self
.RO
.deploy(nsr_id
, target
)
1139 self
.logger
.debug("RO return > {}".format(desc
))
1140 action_id
= desc
["action_id"]
1141 await self
._wait
_ng
_ro
(
1142 nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
1147 "_admin.deployed.RO.operational-status": "running",
1148 "detailed-status": " ".join(stage
),
1150 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1151 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1152 self
._write
_op
_status
(nslcmop_id
, stage
)
1154 logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
)
1158 async def _wait_ng_ro(
1167 detailed_status_old
= None
1169 start_time
= start_time
or time()
1170 while time() <= start_time
+ timeout
:
1171 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
1172 self
.logger
.debug("Wait NG RO > {}".format(desc_status
))
1173 if desc_status
["status"] == "FAILED":
1174 raise NgRoException(desc_status
["details"])
1175 elif desc_status
["status"] == "BUILD":
1177 stage
[2] = "VIM: ({})".format(desc_status
["details"])
1178 elif desc_status
["status"] == "DONE":
1180 stage
[2] = "Deployed at VIM"
1183 assert False, "ROclient.check_ns_status returns unknown {}".format(
1184 desc_status
["status"]
1186 if stage
and nslcmop_id
and stage
[2] != detailed_status_old
:
1187 detailed_status_old
= stage
[2]
1188 db_nsr_update
["detailed-status"] = " ".join(stage
)
1189 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1190 self
._write
_op
_status
(nslcmop_id
, stage
)
1191 await asyncio
.sleep(15, loop
=self
.loop
)
1192 else: # timeout_ns_deploy
1193 raise NgRoException("Timeout waiting ns to deploy")
1195 async def _terminate_ng_ro(
1196 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
1201 start_deploy
= time()
1208 "action_id": nslcmop_id
,
1210 desc
= await self
.RO
.deploy(nsr_id
, target
)
1211 action_id
= desc
["action_id"]
1212 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1213 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1216 + "ns terminate action at RO. action_id={}".format(action_id
)
1220 delete_timeout
= 20 * 60 # 20 minutes
1221 await self
._wait
_ng
_ro
(
1222 nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
1225 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1226 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1228 await self
.RO
.delete(nsr_id
)
1229 except Exception as e
:
1230 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1231 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1232 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1233 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1235 logging_text
+ "RO_action_id={} already deleted".format(action_id
)
1237 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1238 failed_detail
.append("delete conflict: {}".format(e
))
1241 + "RO_action_id={} delete conflict: {}".format(action_id
, e
)
1244 failed_detail
.append("delete error: {}".format(e
))
1247 + "RO_action_id={} delete error: {}".format(action_id
, e
)
1251 stage
[2] = "Error deleting from VIM"
1253 stage
[2] = "Deleted from VIM"
1254 db_nsr_update
["detailed-status"] = " ".join(stage
)
1255 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1256 self
._write
_op
_status
(nslcmop_id
, stage
)
1259 raise LcmException("; ".join(failed_detail
))
1262 async def instantiate_RO(
1276 :param logging_text: preffix text to use at logging
1277 :param nsr_id: nsr identity
1278 :param nsd: database content of ns descriptor
1279 :param db_nsr: database content of ns record
1280 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1282 :param db_vnfds: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1283 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1284 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1285 :return: None or exception
1288 start_deploy
= time()
1289 ns_params
= db_nslcmop
.get("operationParams")
1290 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1291 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1293 timeout_ns_deploy
= self
.timeout
.get(
1294 "ns_deploy", self
.timeout_ns_deploy
1297 # Check for and optionally request placement optimization. Database will be updated if placement activated
1298 stage
[2] = "Waiting for Placement."
1299 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1300 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1301 for vnfr
in db_vnfrs
.values():
1302 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1305 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1307 return await self
._instantiate
_ng
_ro
(
1320 except Exception as e
:
1321 stage
[2] = "ERROR deploying at VIM"
1322 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1324 "Error deploying at VIM {}".format(e
),
1325 exc_info
=not isinstance(
1328 ROclient
.ROClientException
,
1337 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1339 Wait for kdu to be up, get ip address
1340 :param logging_text: prefix use for logging
1347 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1350 while nb_tries
< 360:
1351 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1355 for x
in get_iterable(db_vnfr
, "kdur")
1356 if x
.get("kdu-name") == kdu_name
1362 "Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
)
1364 if kdur
.get("status"):
1365 if kdur
["status"] in ("READY", "ENABLED"):
1366 return kdur
.get("ip-address")
1369 "target KDU={} is in error state".format(kdu_name
)
1372 await asyncio
.sleep(10, loop
=self
.loop
)
1374 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1376 async def wait_vm_up_insert_key_ro(
1377 self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None
1380 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1381 :param logging_text: prefix use for logging
1386 :param pub_key: public ssh key to inject, None to skip
1387 :param user: user to apply the public ssh key
1391 self
.logger
.debug(logging_text
+ "Starting wait_vm_up_insert_key_ro")
1395 target_vdu_id
= None
1401 if ro_retries
>= 360: # 1 hour
1403 "Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
)
1406 await asyncio
.sleep(10, loop
=self
.loop
)
1409 if not target_vdu_id
:
1410 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1412 if not vdu_id
: # for the VNF case
1413 if db_vnfr
.get("status") == "ERROR":
1415 "Cannot inject ssh-key because target VNF is in error state"
1417 ip_address
= db_vnfr
.get("ip-address")
1423 for x
in get_iterable(db_vnfr
, "vdur")
1424 if x
.get("ip-address") == ip_address
1432 for x
in get_iterable(db_vnfr
, "vdur")
1433 if x
.get("vdu-id-ref") == vdu_id
1434 and x
.get("count-index") == vdu_index
1440 not vdur
and len(db_vnfr
.get("vdur", ())) == 1
1441 ): # If only one, this should be the target vdu
1442 vdur
= db_vnfr
["vdur"][0]
1445 "Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(
1446 vnfr_id
, vdu_id
, vdu_index
1449 # New generation RO stores information at "vim_info"
1452 if vdur
.get("vim_info"):
1454 t
for t
in vdur
["vim_info"]
1455 ) # there should be only one key
1456 ng_ro_status
= vdur
["vim_info"][target_vim
].get("vim_status")
1458 vdur
.get("pdu-type")
1459 or vdur
.get("status") == "ACTIVE"
1460 or ng_ro_status
== "ACTIVE"
1462 ip_address
= vdur
.get("ip-address")
1465 target_vdu_id
= vdur
["vdu-id-ref"]
1466 elif vdur
.get("status") == "ERROR" or ng_ro_status
== "ERROR":
1468 "Cannot inject ssh-key because target VM is in error state"
1471 if not target_vdu_id
:
1474 # inject public key into machine
1475 if pub_key
and user
:
1476 self
.logger
.debug(logging_text
+ "Inserting RO key")
1477 self
.logger
.debug("SSH > PubKey > {}".format(pub_key
))
1478 if vdur
.get("pdu-type"):
1479 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1482 ro_vm_id
= "{}-{}".format(
1483 db_vnfr
["member-vnf-index-ref"], target_vdu_id
1484 ) # TODO add vdu_index
1488 "action": "inject_ssh_key",
1492 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdur
["id"]}]}],
1494 desc
= await self
.RO
.deploy(nsr_id
, target
)
1495 action_id
= desc
["action_id"]
1496 await self
._wait
_ng
_ro
(nsr_id
, action_id
, timeout
=600)
1499 # wait until NS is deployed at RO
1501 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1502 ro_nsr_id
= deep_get(
1503 db_nsrs
, ("_admin", "deployed", "RO", "nsr_id")
1507 result_dict
= await self
.RO
.create_action(
1509 item_id_name
=ro_nsr_id
,
1511 "add_public_key": pub_key
,
1516 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1517 if not result_dict
or not isinstance(result_dict
, dict):
1519 "Unknown response from RO when injecting key"
1521 for result
in result_dict
.values():
1522 if result
.get("vim_result") == 200:
1525 raise ROclient
.ROClientException(
1526 "error injecting key: {}".format(
1527 result
.get("description")
1531 except NgRoException
as e
:
1533 "Reaching max tries injecting key. Error: {}".format(e
)
1535 except ROclient
.ROClientException
as e
:
1539 + "error injecting key: {}. Retrying until {} seconds".format(
1546 "Reaching max tries injecting key. Error: {}".format(e
)
1553 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1555 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1557 my_vca
= vca_deployed_list
[vca_index
]
1558 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1559 # vdu or kdu: no dependencies
1563 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1564 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1565 configuration_status_list
= db_nsr
["configurationStatus"]
1566 for index
, vca_deployed
in enumerate(configuration_status_list
):
1567 if index
== vca_index
:
1570 if not my_vca
.get("member-vnf-index") or (
1571 vca_deployed
.get("member-vnf-index")
1572 == my_vca
.get("member-vnf-index")
1574 internal_status
= configuration_status_list
[index
].get("status")
1575 if internal_status
== "READY":
1577 elif internal_status
== "BROKEN":
1579 "Configuration aborted because dependent charm/s has failed"
1584 # no dependencies, return
1586 await asyncio
.sleep(10)
1589 raise LcmException("Configuration aborted because dependent charm/s timeout")
1591 def get_vca_id(self
, db_vnfr
: dict, db_nsr
: dict):
1594 vca_id
= deep_get(db_vnfr
, ("vca-id",))
1596 vim_account_id
= deep_get(db_nsr
, ("instantiate_params", "vimAccountId"))
1597 vca_id
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("vca")
1600 async def instantiate_N2VC(
1617 ee_config_descriptor
,
1619 nsr_id
= db_nsr
["_id"]
1620 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1621 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1622 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1623 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1625 "collection": "nsrs",
1626 "filter": {"_id": nsr_id
},
1627 "path": db_update_entry
,
1633 element_under_configuration
= nsr_id
1637 vnfr_id
= db_vnfr
["_id"]
1638 osm_config
["osm"]["vnf_id"] = vnfr_id
1640 namespace
= "{nsi}.{ns}".format(nsi
=nsi_id
if nsi_id
else "", ns
=nsr_id
)
1642 if vca_type
== "native_charm":
1645 index_number
= vdu_index
or 0
1648 element_type
= "VNF"
1649 element_under_configuration
= vnfr_id
1650 namespace
+= ".{}-{}".format(vnfr_id
, index_number
)
1652 namespace
+= ".{}-{}".format(vdu_id
, index_number
)
1653 element_type
= "VDU"
1654 element_under_configuration
= "{}-{}".format(vdu_id
, index_number
)
1655 osm_config
["osm"]["vdu_id"] = vdu_id
1657 namespace
+= ".{}".format(kdu_name
)
1658 element_type
= "KDU"
1659 element_under_configuration
= kdu_name
1660 osm_config
["osm"]["kdu_name"] = kdu_name
1663 if base_folder
["pkg-dir"]:
1664 artifact_path
= "{}/{}/{}/{}".format(
1665 base_folder
["folder"],
1666 base_folder
["pkg-dir"],
1668 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1673 artifact_path
= "{}/Scripts/{}/{}/".format(
1674 base_folder
["folder"],
1676 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1681 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1683 # get initial_config_primitive_list that applies to this element
1684 initial_config_primitive_list
= config_descriptor
.get(
1685 "initial-config-primitive"
1689 "Initial config primitive list > {}".format(
1690 initial_config_primitive_list
1694 # add config if not present for NS charm
1695 ee_descriptor_id
= ee_config_descriptor
.get("id")
1696 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1697 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1698 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1702 "Initial config primitive list #2 > {}".format(
1703 initial_config_primitive_list
1706 # n2vc_redesign STEP 3.1
1707 # find old ee_id if exists
1708 ee_id
= vca_deployed
.get("ee_id")
1710 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1711 # create or register execution environment in VCA
1712 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1714 self
._write
_configuration
_status
(
1716 vca_index
=vca_index
,
1718 element_under_configuration
=element_under_configuration
,
1719 element_type
=element_type
,
1722 step
= "create execution environment"
1723 self
.logger
.debug(logging_text
+ step
)
1727 if vca_type
== "k8s_proxy_charm":
1728 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1729 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1730 namespace
=namespace
,
1731 artifact_path
=artifact_path
,
1735 elif vca_type
== "helm" or vca_type
== "helm-v3":
1736 ee_id
, credentials
= await self
.vca_map
[
1738 ].create_execution_environment(
1739 namespace
=namespace
,
1743 artifact_path
=artifact_path
,
1747 ee_id
, credentials
= await self
.vca_map
[
1749 ].create_execution_environment(
1750 namespace
=namespace
,
1756 elif vca_type
== "native_charm":
1757 step
= "Waiting to VM being up and getting IP address"
1758 self
.logger
.debug(logging_text
+ step
)
1759 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1768 credentials
= {"hostname": rw_mgmt_ip
}
1770 username
= deep_get(
1771 config_descriptor
, ("config-access", "ssh-access", "default-user")
1773 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1774 # merged. Meanwhile let's get username from initial-config-primitive
1775 if not username
and initial_config_primitive_list
:
1776 for config_primitive
in initial_config_primitive_list
:
1777 for param
in config_primitive
.get("parameter", ()):
1778 if param
["name"] == "ssh-username":
1779 username
= param
["value"]
1783 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1784 "'config-access.ssh-access.default-user'"
1786 credentials
["username"] = username
1787 # n2vc_redesign STEP 3.2
1789 self
._write
_configuration
_status
(
1791 vca_index
=vca_index
,
1792 status
="REGISTERING",
1793 element_under_configuration
=element_under_configuration
,
1794 element_type
=element_type
,
1797 step
= "register execution environment {}".format(credentials
)
1798 self
.logger
.debug(logging_text
+ step
)
1799 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1800 credentials
=credentials
,
1801 namespace
=namespace
,
1806 # for compatibility with MON/POL modules, the need model and application name at database
1807 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1808 ee_id_parts
= ee_id
.split(".")
1809 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1810 if len(ee_id_parts
) >= 2:
1811 model_name
= ee_id_parts
[0]
1812 application_name
= ee_id_parts
[1]
1813 db_nsr_update
[db_update_entry
+ "model"] = model_name
1814 db_nsr_update
[db_update_entry
+ "application"] = application_name
1816 # n2vc_redesign STEP 3.3
1817 step
= "Install configuration Software"
1819 self
._write
_configuration
_status
(
1821 vca_index
=vca_index
,
1822 status
="INSTALLING SW",
1823 element_under_configuration
=element_under_configuration
,
1824 element_type
=element_type
,
1825 other_update
=db_nsr_update
,
1828 # TODO check if already done
1829 self
.logger
.debug(logging_text
+ step
)
1831 if vca_type
== "native_charm":
1832 config_primitive
= next(
1833 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1836 if config_primitive
:
1837 config
= self
._map
_primitive
_params
(
1838 config_primitive
, {}, deploy_params
1841 if vca_type
== "lxc_proxy_charm":
1842 if element_type
== "NS":
1843 num_units
= db_nsr
.get("config-units") or 1
1844 elif element_type
== "VNF":
1845 num_units
= db_vnfr
.get("config-units") or 1
1846 elif element_type
== "VDU":
1847 for v
in db_vnfr
["vdur"]:
1848 if vdu_id
== v
["vdu-id-ref"]:
1849 num_units
= v
.get("config-units") or 1
1851 if vca_type
!= "k8s_proxy_charm":
1852 await self
.vca_map
[vca_type
].install_configuration_sw(
1854 artifact_path
=artifact_path
,
1857 num_units
=num_units
,
1862 # write in db flag of configuration_sw already installed
1864 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1867 # add relations for this VCA (wait for other peers related with this VCA)
1868 await self
._add
_vca
_relations
(
1869 logging_text
=logging_text
,
1872 vca_index
=vca_index
,
1875 # if SSH access is required, then get execution environment SSH public
1876 # if native charm we have waited already to VM be UP
1877 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1880 # self.logger.debug("get ssh key block")
1882 config_descriptor
, ("config-access", "ssh-access", "required")
1884 # self.logger.debug("ssh key needed")
1885 # Needed to inject a ssh key
1888 ("config-access", "ssh-access", "default-user"),
1890 step
= "Install configuration Software, getting public ssh key"
1891 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1892 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1895 step
= "Insert public key into VM user={} ssh_key={}".format(
1899 # self.logger.debug("no need to get ssh key")
1900 step
= "Waiting to VM being up and getting IP address"
1901 self
.logger
.debug(logging_text
+ step
)
1903 # n2vc_redesign STEP 5.1
1904 # wait for RO (ip-address) Insert pub_key into VM
1907 rw_mgmt_ip
= await self
.wait_kdu_up(
1908 logging_text
, nsr_id
, vnfr_id
, kdu_name
1911 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1921 rw_mgmt_ip
= None # This is for a NS configuration
1923 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1925 # store rw_mgmt_ip in deploy params for later replacement
1926 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1928 # n2vc_redesign STEP 6 Execute initial config primitive
1929 step
= "execute initial config primitive"
1931 # wait for dependent primitives execution (NS -> VNF -> VDU)
1932 if initial_config_primitive_list
:
1933 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1935 # stage, in function of element type: vdu, kdu, vnf or ns
1936 my_vca
= vca_deployed_list
[vca_index
]
1937 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1939 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1940 elif my_vca
.get("member-vnf-index"):
1942 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1945 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1947 self
._write
_configuration
_status
(
1948 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1951 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1953 check_if_terminated_needed
= True
1954 for initial_config_primitive
in initial_config_primitive_list
:
1955 # adding information on the vca_deployed if it is a NS execution environment
1956 if not vca_deployed
["member-vnf-index"]:
1957 deploy_params
["ns_config_info"] = json
.dumps(
1958 self
._get
_ns
_config
_info
(nsr_id
)
1960 # TODO check if already done
1961 primitive_params_
= self
._map
_primitive
_params
(
1962 initial_config_primitive
, {}, deploy_params
1965 step
= "execute primitive '{}' params '{}'".format(
1966 initial_config_primitive
["name"], primitive_params_
1968 self
.logger
.debug(logging_text
+ step
)
1969 await self
.vca_map
[vca_type
].exec_primitive(
1971 primitive_name
=initial_config_primitive
["name"],
1972 params_dict
=primitive_params_
,
1977 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1978 if check_if_terminated_needed
:
1979 if config_descriptor
.get("terminate-config-primitive"):
1981 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
1983 check_if_terminated_needed
= False
1985 # TODO register in database that primitive is done
1987 # STEP 7 Configure metrics
1988 if vca_type
== "helm" or vca_type
== "helm-v3":
1989 prometheus_jobs
= await self
.extract_prometheus_scrape_jobs(
1991 artifact_path
=artifact_path
,
1992 ee_config_descriptor
=ee_config_descriptor
,
1995 target_ip
=rw_mgmt_ip
,
2001 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
2004 for job
in prometheus_jobs
:
2008 "job_name": job
["job_name"]
2015 step
= "instantiated at VCA"
2016 self
.logger
.debug(logging_text
+ step
)
2018 self
._write
_configuration
_status
(
2019 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
2022 except Exception as e
: # TODO not use Exception but N2VC exception
2023 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
2025 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
2028 "Exception while {} : {}".format(step
, e
), exc_info
=True
2030 self
._write
_configuration
_status
(
2031 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
2033 raise LcmException("{} {}".format(step
, e
)) from e
2035 def _write_ns_status(
2039 current_operation
: str,
2040 current_operation_id
: str,
2041 error_description
: str = None,
2042 error_detail
: str = None,
2043 other_update
: dict = None,
2046 Update db_nsr fields.
2049 :param current_operation:
2050 :param current_operation_id:
2051 :param error_description:
2052 :param error_detail:
2053 :param other_update: Other required changes at database if provided, will be cleared
2057 db_dict
= other_update
or {}
2060 ] = current_operation_id
# for backward compatibility
2061 db_dict
["_admin.current-operation"] = current_operation_id
2062 db_dict
["_admin.operation-type"] = (
2063 current_operation
if current_operation
!= "IDLE" else None
2065 db_dict
["currentOperation"] = current_operation
2066 db_dict
["currentOperationID"] = current_operation_id
2067 db_dict
["errorDescription"] = error_description
2068 db_dict
["errorDetail"] = error_detail
2071 db_dict
["nsState"] = ns_state
2072 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2073 except DbException
as e
:
2074 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2076 def _write_op_status(
2080 error_message
: str = None,
2081 queuePosition
: int = 0,
2082 operation_state
: str = None,
2083 other_update
: dict = None,
2086 db_dict
= other_update
or {}
2087 db_dict
["queuePosition"] = queuePosition
2088 if isinstance(stage
, list):
2089 db_dict
["stage"] = stage
[0]
2090 db_dict
["detailed-status"] = " ".join(stage
)
2091 elif stage
is not None:
2092 db_dict
["stage"] = str(stage
)
2094 if error_message
is not None:
2095 db_dict
["errorMessage"] = error_message
2096 if operation_state
is not None:
2097 db_dict
["operationState"] = operation_state
2098 db_dict
["statusEnteredTime"] = time()
2099 self
.update_db_2("nslcmops", op_id
, db_dict
)
2100 except DbException
as e
:
2102 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2105 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2107 nsr_id
= db_nsr
["_id"]
2108 # configurationStatus
2109 config_status
= db_nsr
.get("configurationStatus")
2112 "configurationStatus.{}.status".format(index
): status
2113 for index
, v
in enumerate(config_status
)
2117 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2119 except DbException
as e
:
2121 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2124 def _write_configuration_status(
2129 element_under_configuration
: str = None,
2130 element_type
: str = None,
2131 other_update
: dict = None,
2134 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2135 # .format(vca_index, status))
2138 db_path
= "configurationStatus.{}.".format(vca_index
)
2139 db_dict
= other_update
or {}
2141 db_dict
[db_path
+ "status"] = status
2142 if element_under_configuration
:
2144 db_path
+ "elementUnderConfiguration"
2145 ] = element_under_configuration
2147 db_dict
[db_path
+ "elementType"] = element_type
2148 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2149 except DbException
as e
:
2151 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2152 status
, nsr_id
, vca_index
, e
2156 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2158 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2159 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2160 Database is used because the result can be obtained from a different LCM worker in case of HA.
2161 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2162 :param db_nslcmop: database content of nslcmop
2163 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2164 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2165 computed 'vim-account-id'
2168 nslcmop_id
= db_nslcmop
["_id"]
2169 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2170 if placement_engine
== "PLA":
2172 logging_text
+ "Invoke and wait for placement optimization"
2174 await self
.msg
.aiowrite(
2175 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2177 db_poll_interval
= 5
2178 wait
= db_poll_interval
* 10
2180 while not pla_result
and wait
>= 0:
2181 await asyncio
.sleep(db_poll_interval
)
2182 wait
-= db_poll_interval
2183 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2184 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2188 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2191 for pla_vnf
in pla_result
["vnf"]:
2192 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2193 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2198 {"_id": vnfr
["_id"]},
2199 {"vim-account-id": pla_vnf
["vimAccountId"]},
2202 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2205 def update_nsrs_with_pla_result(self
, params
):
2207 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2209 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2211 except Exception as e
:
2212 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2214 async def instantiate(self
, nsr_id
, nslcmop_id
):
2217 :param nsr_id: ns instance to deploy
2218 :param nslcmop_id: operation to run
2222 # Try to lock HA task here
2223 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2224 if not task_is_locked_by_me
:
2226 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2230 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2231 self
.logger
.debug(logging_text
+ "Enter")
2233 # get all needed from database
2235 # database nsrs record
2238 # database nslcmops record
2241 # update operation on nsrs
2243 # update operation on nslcmops
2244 db_nslcmop_update
= {}
2246 nslcmop_operation_state
= None
2247 db_vnfrs
= {} # vnf's info indexed by member-index
2249 tasks_dict_info
= {} # from task to info text
2253 "Stage 1/5: preparation of the environment.",
2254 "Waiting for previous operations to terminate.",
2257 # ^ stage, step, VIM progress
2259 # wait for any previous tasks in process
2260 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2262 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2263 stage
[1] = "Reading from database."
2264 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2265 db_nsr_update
["detailed-status"] = "creating"
2266 db_nsr_update
["operational-status"] = "init"
2267 self
._write
_ns
_status
(
2269 ns_state
="BUILDING",
2270 current_operation
="INSTANTIATING",
2271 current_operation_id
=nslcmop_id
,
2272 other_update
=db_nsr_update
,
2274 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2276 # read from db: operation
2277 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2278 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2279 if db_nslcmop
["operationParams"].get("additionalParamsForVnf"):
2280 db_nslcmop
["operationParams"]["additionalParamsForVnf"] = json
.loads(
2281 db_nslcmop
["operationParams"]["additionalParamsForVnf"]
2283 ns_params
= db_nslcmop
.get("operationParams")
2284 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2285 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2287 timeout_ns_deploy
= self
.timeout
.get(
2288 "ns_deploy", self
.timeout_ns_deploy
2292 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2293 self
.logger
.debug(logging_text
+ stage
[1])
2294 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2295 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2296 self
.logger
.debug(logging_text
+ stage
[1])
2297 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2298 self
.fs
.sync(db_nsr
["nsd-id"])
2300 # nsr_name = db_nsr["name"] # TODO short-name??
2302 # read from db: vnf's of this ns
2303 stage
[1] = "Getting vnfrs from db."
2304 self
.logger
.debug(logging_text
+ stage
[1])
2305 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2307 # read from db: vnfd's for every vnf
2308 db_vnfds
= [] # every vnfd data
2310 # for each vnf in ns, read vnfd
2311 for vnfr
in db_vnfrs_list
:
2312 if vnfr
.get("kdur"):
2314 for kdur
in vnfr
["kdur"]:
2315 if kdur
.get("additionalParams"):
2316 kdur
["additionalParams"] = json
.loads(kdur
["additionalParams"])
2317 kdur_list
.append(kdur
)
2318 vnfr
["kdur"] = kdur_list
2320 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2321 vnfd_id
= vnfr
["vnfd-id"]
2322 vnfd_ref
= vnfr
["vnfd-ref"]
2323 self
.fs
.sync(vnfd_id
)
2325 # if we haven't this vnfd, read it from db
2326 if vnfd_id
not in db_vnfds
:
2328 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2331 self
.logger
.debug(logging_text
+ stage
[1])
2332 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2335 db_vnfds
.append(vnfd
)
2337 # Get or generates the _admin.deployed.VCA list
2338 vca_deployed_list
= None
2339 if db_nsr
["_admin"].get("deployed"):
2340 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2341 if vca_deployed_list
is None:
2342 vca_deployed_list
= []
2343 configuration_status_list
= []
2344 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2345 db_nsr_update
["configurationStatus"] = configuration_status_list
2346 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2347 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2348 elif isinstance(vca_deployed_list
, dict):
2349 # maintain backward compatibility. Change a dict to list at database
2350 vca_deployed_list
= list(vca_deployed_list
.values())
2351 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2352 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2355 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2357 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2358 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2360 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2361 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2362 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2364 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2367 # n2vc_redesign STEP 2 Deploy Network Scenario
2368 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2369 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2371 stage
[1] = "Deploying KDUs."
2372 # self.logger.debug(logging_text + "Before deploy_kdus")
2373 # Call to deploy_kdus in case exists the "vdu:kdu" param
2374 await self
.deploy_kdus(
2375 logging_text
=logging_text
,
2377 nslcmop_id
=nslcmop_id
,
2380 task_instantiation_info
=tasks_dict_info
,
2383 stage
[1] = "Getting VCA public key."
2384 # n2vc_redesign STEP 1 Get VCA public ssh-key
2385 # feature 1429. Add n2vc public key to needed VMs
2386 n2vc_key
= self
.n2vc
.get_public_key()
2387 n2vc_key_list
= [n2vc_key
]
2388 if self
.vca_config
.get("public_key"):
2389 n2vc_key_list
.append(self
.vca_config
["public_key"])
2391 stage
[1] = "Deploying NS at VIM."
2392 task_ro
= asyncio
.ensure_future(
2393 self
.instantiate_RO(
2394 logging_text
=logging_text
,
2398 db_nslcmop
=db_nslcmop
,
2401 n2vc_key_list
=n2vc_key_list
,
2405 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2406 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2408 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2409 stage
[1] = "Deploying Execution Environments."
2410 self
.logger
.debug(logging_text
+ stage
[1])
2412 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2413 for vnf_profile
in get_vnf_profiles(nsd
):
2414 vnfd_id
= vnf_profile
["vnfd-id"]
2415 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2416 member_vnf_index
= str(vnf_profile
["id"])
2417 db_vnfr
= db_vnfrs
[member_vnf_index
]
2418 base_folder
= vnfd
["_admin"]["storage"]
2424 # Get additional parameters
2425 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2426 if db_vnfr
.get("additionalParamsForVnf"):
2427 deploy_params
.update(
2428 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2431 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2432 if descriptor_config
:
2434 logging_text
=logging_text
2435 + "member_vnf_index={} ".format(member_vnf_index
),
2438 nslcmop_id
=nslcmop_id
,
2444 member_vnf_index
=member_vnf_index
,
2445 vdu_index
=vdu_index
,
2447 deploy_params
=deploy_params
,
2448 descriptor_config
=descriptor_config
,
2449 base_folder
=base_folder
,
2450 task_instantiation_info
=tasks_dict_info
,
2454 # Deploy charms for each VDU that supports one.
2455 for vdud
in get_vdu_list(vnfd
):
2457 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2458 vdur
= find_in_list(
2459 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2462 if vdur
.get("additionalParams"):
2463 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2465 deploy_params_vdu
= deploy_params
2466 deploy_params_vdu
["OSM"] = get_osm_params(
2467 db_vnfr
, vdu_id
, vdu_count_index
=0
2469 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2471 self
.logger
.debug("VDUD > {}".format(vdud
))
2473 "Descriptor config > {}".format(descriptor_config
)
2475 if descriptor_config
:
2478 for vdu_index
in range(vdud_count
):
2479 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2481 logging_text
=logging_text
2482 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2483 member_vnf_index
, vdu_id
, vdu_index
2487 nslcmop_id
=nslcmop_id
,
2493 member_vnf_index
=member_vnf_index
,
2494 vdu_index
=vdu_index
,
2496 deploy_params
=deploy_params_vdu
,
2497 descriptor_config
=descriptor_config
,
2498 base_folder
=base_folder
,
2499 task_instantiation_info
=tasks_dict_info
,
2502 for kdud
in get_kdu_list(vnfd
):
2503 kdu_name
= kdud
["name"]
2504 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2505 if descriptor_config
:
2510 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2512 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2513 if kdur
.get("additionalParams"):
2514 deploy_params_kdu
= parse_yaml_strings(
2515 kdur
["additionalParams"]
2519 logging_text
=logging_text
,
2522 nslcmop_id
=nslcmop_id
,
2528 member_vnf_index
=member_vnf_index
,
2529 vdu_index
=vdu_index
,
2531 deploy_params
=deploy_params_kdu
,
2532 descriptor_config
=descriptor_config
,
2533 base_folder
=base_folder
,
2534 task_instantiation_info
=tasks_dict_info
,
2538 # Check if this NS has a charm configuration
2539 descriptor_config
= nsd
.get("ns-configuration")
2540 if descriptor_config
and descriptor_config
.get("juju"):
2543 member_vnf_index
= None
2549 # Get additional parameters
2550 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2551 if db_nsr
.get("additionalParamsForNs"):
2552 deploy_params
.update(
2553 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2555 base_folder
= nsd
["_admin"]["storage"]
2557 logging_text
=logging_text
,
2560 nslcmop_id
=nslcmop_id
,
2566 member_vnf_index
=member_vnf_index
,
2567 vdu_index
=vdu_index
,
2569 deploy_params
=deploy_params
,
2570 descriptor_config
=descriptor_config
,
2571 base_folder
=base_folder
,
2572 task_instantiation_info
=tasks_dict_info
,
2576 # rest of staff will be done at finally
2579 ROclient
.ROClientException
,
2585 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2588 except asyncio
.CancelledError
:
2590 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2592 exc
= "Operation was cancelled"
2593 except Exception as e
:
2594 exc
= traceback
.format_exc()
2595 self
.logger
.critical(
2596 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2601 error_list
.append(str(exc
))
2603 # wait for pending tasks
2605 stage
[1] = "Waiting for instantiate pending tasks."
2606 self
.logger
.debug(logging_text
+ stage
[1])
2607 error_list
+= await self
._wait
_for
_tasks
(
2615 stage
[1] = stage
[2] = ""
2616 except asyncio
.CancelledError
:
2617 error_list
.append("Cancelled")
2618 # TODO cancel all tasks
2619 except Exception as exc
:
2620 error_list
.append(str(exc
))
2622 # update operation-status
2623 db_nsr_update
["operational-status"] = "running"
2624 # let's begin with VCA 'configured' status (later we can change it)
2625 db_nsr_update
["config-status"] = "configured"
2626 for task
, task_name
in tasks_dict_info
.items():
2627 if not task
.done() or task
.cancelled() or task
.exception():
2628 if task_name
.startswith(self
.task_name_deploy_vca
):
2629 # A N2VC task is pending
2630 db_nsr_update
["config-status"] = "failed"
2632 # RO or KDU task is pending
2633 db_nsr_update
["operational-status"] = "failed"
2635 # update status at database
2637 error_detail
= ". ".join(error_list
)
2638 self
.logger
.error(logging_text
+ error_detail
)
2639 error_description_nslcmop
= "{} Detail: {}".format(
2640 stage
[0], error_detail
2642 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2643 nslcmop_id
, stage
[0]
2646 db_nsr_update
["detailed-status"] = (
2647 error_description_nsr
+ " Detail: " + error_detail
2649 db_nslcmop_update
["detailed-status"] = error_detail
2650 nslcmop_operation_state
= "FAILED"
2654 error_description_nsr
= error_description_nslcmop
= None
2656 db_nsr_update
["detailed-status"] = "Done"
2657 db_nslcmop_update
["detailed-status"] = "Done"
2658 nslcmop_operation_state
= "COMPLETED"
2661 self
._write
_ns
_status
(
2664 current_operation
="IDLE",
2665 current_operation_id
=None,
2666 error_description
=error_description_nsr
,
2667 error_detail
=error_detail
,
2668 other_update
=db_nsr_update
,
2670 self
._write
_op
_status
(
2673 error_message
=error_description_nslcmop
,
2674 operation_state
=nslcmop_operation_state
,
2675 other_update
=db_nslcmop_update
,
2678 if nslcmop_operation_state
:
2680 await self
.msg
.aiowrite(
2685 "nslcmop_id": nslcmop_id
,
2686 "operationState": nslcmop_operation_state
,
2690 except Exception as e
:
2692 logging_text
+ "kafka_write notification Exception {}".format(e
)
2695 self
.logger
.debug(logging_text
+ "Exit")
2696 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2698 def _get_vnfd(self
, vnfd_id
: str, cached_vnfds
: Dict
[str, Any
]):
2699 if vnfd_id
not in cached_vnfds
:
2700 cached_vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
2701 return cached_vnfds
[vnfd_id
]
2703 def _get_vnfr(self
, nsr_id
: str, vnf_profile_id
: str, cached_vnfrs
: Dict
[str, Any
]):
2704 if vnf_profile_id
not in cached_vnfrs
:
2705 cached_vnfrs
[vnf_profile_id
] = self
.db
.get_one(
2708 "member-vnf-index-ref": vnf_profile_id
,
2709 "nsr-id-ref": nsr_id
,
2712 return cached_vnfrs
[vnf_profile_id
]
2714 def _is_deployed_vca_in_relation(
2715 self
, vca
: DeployedVCA
, relation
: Relation
2718 for endpoint
in (relation
.provider
, relation
.requirer
):
2719 if endpoint
["kdu-resource-profile-id"]:
2722 vca
.vnf_profile_id
== endpoint
.vnf_profile_id
2723 and vca
.vdu_profile_id
== endpoint
.vdu_profile_id
2724 and vca
.execution_environment_ref
== endpoint
.execution_environment_ref
2730 def _update_ee_relation_data_with_implicit_data(
2731 self
, nsr_id
, nsd
, ee_relation_data
, cached_vnfds
, vnf_profile_id
: str = None
2733 ee_relation_data
= safe_get_ee_relation(
2734 nsr_id
, ee_relation_data
, vnf_profile_id
=vnf_profile_id
2736 ee_relation_level
= EELevel
.get_level(ee_relation_data
)
2737 if (ee_relation_level
in (EELevel
.VNF
, EELevel
.VDU
)) and not ee_relation_data
[
2738 "execution-environment-ref"
2740 vnf_profile
= get_vnf_profile(nsd
, ee_relation_data
["vnf-profile-id"])
2741 vnfd_id
= vnf_profile
["vnfd-id"]
2742 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2745 if ee_relation_level
== EELevel
.VNF
2746 else ee_relation_data
["vdu-profile-id"]
2748 ee
= get_juju_ee_ref(db_vnfd
, entity_id
)
2751 f
"not execution environments found for ee_relation {ee_relation_data}"
2753 ee_relation_data
["execution-environment-ref"] = ee
["id"]
2754 return ee_relation_data
2756 def _get_ns_relations(
2759 nsd
: Dict
[str, Any
],
2761 cached_vnfds
: Dict
[str, Any
],
2762 ) -> List
[Relation
]:
2764 db_ns_relations
= get_ns_configuration_relation_list(nsd
)
2765 for r
in db_ns_relations
:
2766 provider_dict
= None
2767 requirer_dict
= None
2768 if all(key
in r
for key
in ("provider", "requirer")):
2769 provider_dict
= r
["provider"]
2770 requirer_dict
= r
["requirer"]
2771 elif "entities" in r
:
2772 provider_id
= r
["entities"][0]["id"]
2775 "endpoint": r
["entities"][0]["endpoint"],
2777 if provider_id
!= nsd
["id"]:
2778 provider_dict
["vnf-profile-id"] = provider_id
2779 requirer_id
= r
["entities"][1]["id"]
2782 "endpoint": r
["entities"][1]["endpoint"],
2784 if requirer_id
!= nsd
["id"]:
2785 requirer_dict
["vnf-profile-id"] = requirer_id
2787 raise Exception("provider/requirer or entities must be included in the relation.")
2788 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2789 nsr_id
, nsd
, provider_dict
, cached_vnfds
2791 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2792 nsr_id
, nsd
, requirer_dict
, cached_vnfds
2794 provider
= EERelation(relation_provider
)
2795 requirer
= EERelation(relation_requirer
)
2796 relation
= Relation(r
["name"], provider
, requirer
)
2797 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2799 relations
.append(relation
)
2802 def _get_vnf_relations(
2805 nsd
: Dict
[str, Any
],
2807 cached_vnfds
: Dict
[str, Any
],
2808 ) -> List
[Relation
]:
2810 vnf_profile
= get_vnf_profile(nsd
, vca
.vnf_profile_id
)
2811 vnf_profile_id
= vnf_profile
["id"]
2812 vnfd_id
= vnf_profile
["vnfd-id"]
2813 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2814 db_vnf_relations
= get_relation_list(db_vnfd
, vnfd_id
)
2815 for r
in db_vnf_relations
:
2816 provider_dict
= None
2817 requirer_dict
= None
2818 if all(key
in r
for key
in ("provider", "requirer")):
2819 provider_dict
= r
["provider"]
2820 requirer_dict
= r
["requirer"]
2821 elif "entities" in r
:
2822 provider_id
= r
["entities"][0]["id"]
2825 "vnf-profile-id": vnf_profile_id
,
2826 "endpoint": r
["entities"][0]["endpoint"],
2828 if provider_id
!= vnfd_id
:
2829 provider_dict
["vdu-profile-id"] = provider_id
2830 requirer_id
= r
["entities"][1]["id"]
2833 "vnf-profile-id": vnf_profile_id
,
2834 "endpoint": r
["entities"][1]["endpoint"],
2836 if requirer_id
!= vnfd_id
:
2837 requirer_dict
["vdu-profile-id"] = requirer_id
2839 raise Exception("provider/requirer or entities must be included in the relation.")
2840 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2841 nsr_id
, nsd
, provider_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2843 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2844 nsr_id
, nsd
, requirer_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2846 provider
= EERelation(relation_provider
)
2847 requirer
= EERelation(relation_requirer
)
2848 relation
= Relation(r
["name"], provider
, requirer
)
2849 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2851 relations
.append(relation
)
2854 def _get_kdu_resource_data(
2856 ee_relation
: EERelation
,
2857 db_nsr
: Dict
[str, Any
],
2858 cached_vnfds
: Dict
[str, Any
],
2859 ) -> DeployedK8sResource
:
2860 nsd
= get_nsd(db_nsr
)
2861 vnf_profiles
= get_vnf_profiles(nsd
)
2862 vnfd_id
= find_in_list(
2864 lambda vnf_profile
: vnf_profile
["id"] == ee_relation
.vnf_profile_id
,
2866 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2867 kdu_resource_profile
= get_kdu_resource_profile(
2868 db_vnfd
, ee_relation
.kdu_resource_profile_id
2870 kdu_name
= kdu_resource_profile
["kdu-name"]
2871 deployed_kdu
, _
= get_deployed_kdu(
2872 db_nsr
.get("_admin", ()).get("deployed", ()),
2874 ee_relation
.vnf_profile_id
,
2876 deployed_kdu
.update({"resource-name": kdu_resource_profile
["resource-name"]})
2879 def _get_deployed_component(
2881 ee_relation
: EERelation
,
2882 db_nsr
: Dict
[str, Any
],
2883 cached_vnfds
: Dict
[str, Any
],
2884 ) -> DeployedComponent
:
2885 nsr_id
= db_nsr
["_id"]
2886 deployed_component
= None
2887 ee_level
= EELevel
.get_level(ee_relation
)
2888 if ee_level
== EELevel
.NS
:
2889 vca
= get_deployed_vca(db_nsr
, {"vdu_id": None, "member-vnf-index": None})
2891 deployed_component
= DeployedVCA(nsr_id
, vca
)
2892 elif ee_level
== EELevel
.VNF
:
2893 vca
= get_deployed_vca(
2897 "member-vnf-index": ee_relation
.vnf_profile_id
,
2898 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2902 deployed_component
= DeployedVCA(nsr_id
, vca
)
2903 elif ee_level
== EELevel
.VDU
:
2904 vca
= get_deployed_vca(
2907 "vdu_id": ee_relation
.vdu_profile_id
,
2908 "member-vnf-index": ee_relation
.vnf_profile_id
,
2909 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2913 deployed_component
= DeployedVCA(nsr_id
, vca
)
2914 elif ee_level
== EELevel
.KDU
:
2915 kdu_resource_data
= self
._get
_kdu
_resource
_data
(
2916 ee_relation
, db_nsr
, cached_vnfds
2918 if kdu_resource_data
:
2919 deployed_component
= DeployedK8sResource(kdu_resource_data
)
2920 return deployed_component
2922 async def _add_relation(
2926 db_nsr
: Dict
[str, Any
],
2927 cached_vnfds
: Dict
[str, Any
],
2928 cached_vnfrs
: Dict
[str, Any
],
2930 deployed_provider
= self
._get
_deployed
_component
(
2931 relation
.provider
, db_nsr
, cached_vnfds
2933 deployed_requirer
= self
._get
_deployed
_component
(
2934 relation
.requirer
, db_nsr
, cached_vnfds
2938 and deployed_requirer
2939 and deployed_provider
.config_sw_installed
2940 and deployed_requirer
.config_sw_installed
2942 provider_db_vnfr
= (
2944 relation
.provider
.nsr_id
,
2945 relation
.provider
.vnf_profile_id
,
2948 if relation
.provider
.vnf_profile_id
2951 requirer_db_vnfr
= (
2953 relation
.requirer
.nsr_id
,
2954 relation
.requirer
.vnf_profile_id
,
2957 if relation
.requirer
.vnf_profile_id
2960 provider_vca_id
= self
.get_vca_id(provider_db_vnfr
, db_nsr
)
2961 requirer_vca_id
= self
.get_vca_id(requirer_db_vnfr
, db_nsr
)
2962 provider_relation_endpoint
= RelationEndpoint(
2963 deployed_provider
.ee_id
,
2965 relation
.provider
.endpoint
,
2967 requirer_relation_endpoint
= RelationEndpoint(
2968 deployed_requirer
.ee_id
,
2970 relation
.requirer
.endpoint
,
2972 await self
.vca_map
[vca_type
].add_relation(
2973 provider
=provider_relation_endpoint
,
2974 requirer
=requirer_relation_endpoint
,
2976 # remove entry from relations list
2980 async def _add_vca_relations(
2986 timeout
: int = 3600,
2990 # 1. find all relations for this VCA
2991 # 2. wait for other peers related
2995 # STEP 1: find all relations for this VCA
2998 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2999 nsd
= get_nsd(db_nsr
)
3002 deployed_vca_dict
= get_deployed_vca_list(db_nsr
)[vca_index
]
3003 my_vca
= DeployedVCA(nsr_id
, deployed_vca_dict
)
3008 relations
.extend(self
._get
_ns
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
3009 relations
.extend(self
._get
_vnf
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
3011 # if no relations, terminate
3013 self
.logger
.debug(logging_text
+ " No relations")
3016 self
.logger
.debug(logging_text
+ " adding relations {}".format(relations
))
3023 if now
- start
>= timeout
:
3024 self
.logger
.error(logging_text
+ " : timeout adding relations")
3027 # reload nsr from database (we need to update record: _admin.deployed.VCA)
3028 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3030 # for each relation, find the VCA's related
3031 for relation
in relations
.copy():
3032 added
= await self
._add
_relation
(
3040 relations
.remove(relation
)
3043 self
.logger
.debug("Relations added")
3045 await asyncio
.sleep(5.0)
3049 except Exception as e
:
3050 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
3053 async def _install_kdu(
3061 k8s_instance_info
: dict,
3062 k8params
: dict = None,
3068 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
3071 "collection": "nsrs",
3072 "filter": {"_id": nsr_id
},
3073 "path": nsr_db_path
,
3076 if k8s_instance_info
.get("kdu-deployment-name"):
3077 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
3079 kdu_instance
= self
.k8scluster_map
[
3081 ].generate_kdu_instance_name(
3082 db_dict
=db_dict_install
,
3083 kdu_model
=k8s_instance_info
["kdu-model"],
3084 kdu_name
=k8s_instance_info
["kdu-name"],
3087 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3089 await self
.k8scluster_map
[k8sclustertype
].install(
3090 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3091 kdu_model
=k8s_instance_info
["kdu-model"],
3094 db_dict
=db_dict_install
,
3096 kdu_name
=k8s_instance_info
["kdu-name"],
3097 namespace
=k8s_instance_info
["namespace"],
3098 kdu_instance
=kdu_instance
,
3102 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3105 # Obtain services to obtain management service ip
3106 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
3107 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3108 kdu_instance
=kdu_instance
,
3109 namespace
=k8s_instance_info
["namespace"],
3112 # Obtain management service info (if exists)
3113 vnfr_update_dict
= {}
3114 kdu_config
= get_configuration(vnfd
, kdud
["name"])
3116 target_ee_list
= kdu_config
.get("execution-environment-list", [])
3121 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
3124 for service
in kdud
.get("service", [])
3125 if service
.get("mgmt-service")
3127 for mgmt_service
in mgmt_services
:
3128 for service
in services
:
3129 if service
["name"].startswith(mgmt_service
["name"]):
3130 # Mgmt service found, Obtain service ip
3131 ip
= service
.get("external_ip", service
.get("cluster_ip"))
3132 if isinstance(ip
, list) and len(ip
) == 1:
3136 "kdur.{}.ip-address".format(kdu_index
)
3139 # Check if must update also mgmt ip at the vnf
3140 service_external_cp
= mgmt_service
.get(
3141 "external-connection-point-ref"
3143 if service_external_cp
:
3145 deep_get(vnfd
, ("mgmt-interface", "cp"))
3146 == service_external_cp
3148 vnfr_update_dict
["ip-address"] = ip
3153 "external-connection-point-ref", ""
3155 == service_external_cp
,
3158 "kdur.{}.ip-address".format(kdu_index
)
3163 "Mgmt service name: {} not found".format(
3164 mgmt_service
["name"]
3168 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3169 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3171 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3174 and kdu_config
.get("initial-config-primitive")
3175 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3177 initial_config_primitive_list
= kdu_config
.get(
3178 "initial-config-primitive"
3180 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3182 for initial_config_primitive
in initial_config_primitive_list
:
3183 primitive_params_
= self
._map
_primitive
_params
(
3184 initial_config_primitive
, {}, {}
3187 await asyncio
.wait_for(
3188 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3189 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3190 kdu_instance
=kdu_instance
,
3191 primitive_name
=initial_config_primitive
["name"],
3192 params
=primitive_params_
,
3193 db_dict
=db_dict_install
,
3199 except Exception as e
:
3200 # Prepare update db with error and raise exception
3203 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3207 vnfr_data
.get("_id"),
3208 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3211 # ignore to keep original exception
3213 # reraise original error
3218 async def deploy_kdus(
3225 task_instantiation_info
,
3227 # Launch kdus if present in the descriptor
3229 k8scluster_id_2_uuic
= {
3230 "helm-chart-v3": {},
3235 async def _get_cluster_id(cluster_id
, cluster_type
):
3236 nonlocal k8scluster_id_2_uuic
3237 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3238 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3240 # check if K8scluster is creating and wait look if previous tasks in process
3241 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3242 "k8scluster", cluster_id
3245 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3246 task_name
, cluster_id
3248 self
.logger
.debug(logging_text
+ text
)
3249 await asyncio
.wait(task_dependency
, timeout
=3600)
3251 db_k8scluster
= self
.db
.get_one(
3252 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3254 if not db_k8scluster
:
3255 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3257 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3259 if cluster_type
== "helm-chart-v3":
3261 # backward compatibility for existing clusters that have not been initialized for helm v3
3262 k8s_credentials
= yaml
.safe_dump(
3263 db_k8scluster
.get("credentials")
3265 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3266 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3268 db_k8scluster_update
= {}
3269 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3270 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3271 db_k8scluster_update
[
3272 "_admin.helm-chart-v3.created"
3274 db_k8scluster_update
[
3275 "_admin.helm-chart-v3.operationalState"
3278 "k8sclusters", cluster_id
, db_k8scluster_update
3280 except Exception as e
:
3283 + "error initializing helm-v3 cluster: {}".format(str(e
))
3286 "K8s cluster '{}' has not been initialized for '{}'".format(
3287 cluster_id
, cluster_type
3292 "K8s cluster '{}' has not been initialized for '{}'".format(
3293 cluster_id
, cluster_type
3296 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3299 logging_text
+= "Deploy kdus: "
3302 db_nsr_update
= {"_admin.deployed.K8s": []}
3303 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3306 updated_cluster_list
= []
3307 updated_v3_cluster_list
= []
3309 for vnfr_data
in db_vnfrs
.values():
3310 vca_id
= self
.get_vca_id(vnfr_data
, {})
3311 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3312 # Step 0: Prepare and set parameters
3313 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3314 vnfd_id
= vnfr_data
.get("vnfd-id")
3315 vnfd_with_id
= find_in_list(
3316 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3320 for kdud
in vnfd_with_id
["kdu"]
3321 if kdud
["name"] == kdur
["kdu-name"]
3323 namespace
= kdur
.get("k8s-namespace")
3324 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3325 if kdur
.get("helm-chart"):
3326 kdumodel
= kdur
["helm-chart"]
3327 # Default version: helm3, if helm-version is v2 assign v2
3328 k8sclustertype
= "helm-chart-v3"
3329 self
.logger
.debug("kdur: {}".format(kdur
))
3331 kdur
.get("helm-version")
3332 and kdur
.get("helm-version") == "v2"
3334 k8sclustertype
= "helm-chart"
3335 elif kdur
.get("juju-bundle"):
3336 kdumodel
= kdur
["juju-bundle"]
3337 k8sclustertype
= "juju-bundle"
3340 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3341 "juju-bundle. Maybe an old NBI version is running".format(
3342 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3345 # check if kdumodel is a file and exists
3347 vnfd_with_id
= find_in_list(
3348 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3350 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3351 if storage
: # may be not present if vnfd has not artifacts
3352 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3353 if storage
["pkg-dir"]:
3354 filename
= "{}/{}/{}s/{}".format(
3361 filename
= "{}/Scripts/{}s/{}".format(
3366 if self
.fs
.file_exists(
3367 filename
, mode
="file"
3368 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3369 kdumodel
= self
.fs
.path
+ filename
3370 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3372 except Exception: # it is not a file
3375 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3376 step
= "Synchronize repos for k8s cluster '{}'".format(
3379 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3383 k8sclustertype
== "helm-chart"
3384 and cluster_uuid
not in updated_cluster_list
3386 k8sclustertype
== "helm-chart-v3"
3387 and cluster_uuid
not in updated_v3_cluster_list
3389 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3390 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3391 cluster_uuid
=cluster_uuid
3394 if del_repo_list
or added_repo_dict
:
3395 if k8sclustertype
== "helm-chart":
3397 "_admin.helm_charts_added." + item
: None
3398 for item
in del_repo_list
3401 "_admin.helm_charts_added." + item
: name
3402 for item
, name
in added_repo_dict
.items()
3404 updated_cluster_list
.append(cluster_uuid
)
3405 elif k8sclustertype
== "helm-chart-v3":
3407 "_admin.helm_charts_v3_added." + item
: None
3408 for item
in del_repo_list
3411 "_admin.helm_charts_v3_added." + item
: name
3412 for item
, name
in added_repo_dict
.items()
3414 updated_v3_cluster_list
.append(cluster_uuid
)
3416 logging_text
+ "repos synchronized on k8s cluster "
3417 "'{}' to_delete: {}, to_add: {}".format(
3418 k8s_cluster_id
, del_repo_list
, added_repo_dict
3423 {"_id": k8s_cluster_id
},
3429 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3430 vnfr_data
["member-vnf-index-ref"],
3434 k8s_instance_info
= {
3435 "kdu-instance": None,
3436 "k8scluster-uuid": cluster_uuid
,
3437 "k8scluster-type": k8sclustertype
,
3438 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3439 "kdu-name": kdur
["kdu-name"],
3440 "kdu-model": kdumodel
,
3441 "namespace": namespace
,
3442 "kdu-deployment-name": kdu_deployment_name
,
3444 db_path
= "_admin.deployed.K8s.{}".format(index
)
3445 db_nsr_update
[db_path
] = k8s_instance_info
3446 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3447 vnfd_with_id
= find_in_list(
3448 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3450 task
= asyncio
.ensure_future(
3459 k8params
=desc_params
,
3464 self
.lcm_tasks
.register(
3468 "instantiate_KDU-{}".format(index
),
3471 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3477 except (LcmException
, asyncio
.CancelledError
):
3479 except Exception as e
:
3480 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3481 if isinstance(e
, (N2VCException
, DbException
)):
3482 self
.logger
.error(logging_text
+ msg
)
3484 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3485 raise LcmException(msg
)
3488 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3507 task_instantiation_info
,
3510 # launch instantiate_N2VC in a asyncio task and register task object
3511 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3512 # if not found, create one entry and update database
3513 # fill db_nsr._admin.deployed.VCA.<index>
3516 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3518 if "execution-environment-list" in descriptor_config
:
3519 ee_list
= descriptor_config
.get("execution-environment-list", [])
3520 elif "juju" in descriptor_config
:
3521 ee_list
= [descriptor_config
] # ns charms
3522 else: # other types as script are not supported
3525 for ee_item
in ee_list
:
3528 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3529 ee_item
.get("juju"), ee_item
.get("helm-chart")
3532 ee_descriptor_id
= ee_item
.get("id")
3533 if ee_item
.get("juju"):
3534 vca_name
= ee_item
["juju"].get("charm")
3537 if ee_item
["juju"].get("charm") is not None
3540 if ee_item
["juju"].get("cloud") == "k8s":
3541 vca_type
= "k8s_proxy_charm"
3542 elif ee_item
["juju"].get("proxy") is False:
3543 vca_type
= "native_charm"
3544 elif ee_item
.get("helm-chart"):
3545 vca_name
= ee_item
["helm-chart"]
3546 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3549 vca_type
= "helm-v3"
3552 logging_text
+ "skipping non juju neither charm configuration"
3557 for vca_index
, vca_deployed
in enumerate(
3558 db_nsr
["_admin"]["deployed"]["VCA"]
3560 if not vca_deployed
:
3563 vca_deployed
.get("member-vnf-index") == member_vnf_index
3564 and vca_deployed
.get("vdu_id") == vdu_id
3565 and vca_deployed
.get("kdu_name") == kdu_name
3566 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3567 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3571 # not found, create one.
3573 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3576 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3578 target
+= "/kdu/{}".format(kdu_name
)
3580 "target_element": target
,
3581 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3582 "member-vnf-index": member_vnf_index
,
3584 "kdu_name": kdu_name
,
3585 "vdu_count_index": vdu_index
,
3586 "operational-status": "init", # TODO revise
3587 "detailed-status": "", # TODO revise
3588 "step": "initial-deploy", # TODO revise
3590 "vdu_name": vdu_name
,
3592 "ee_descriptor_id": ee_descriptor_id
,
3596 # create VCA and configurationStatus in db
3598 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3599 "configurationStatus.{}".format(vca_index
): dict(),
3601 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3603 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3605 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3606 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3607 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3610 task_n2vc
= asyncio
.ensure_future(
3611 self
.instantiate_N2VC(
3612 logging_text
=logging_text
,
3613 vca_index
=vca_index
,
3619 vdu_index
=vdu_index
,
3620 deploy_params
=deploy_params
,
3621 config_descriptor
=descriptor_config
,
3622 base_folder
=base_folder
,
3623 nslcmop_id
=nslcmop_id
,
3627 ee_config_descriptor
=ee_item
,
3630 self
.lcm_tasks
.register(
3634 "instantiate_N2VC-{}".format(vca_index
),
3637 task_instantiation_info
[
3639 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3640 member_vnf_index
or "", vdu_id
or ""
3644 def _create_nslcmop(nsr_id
, operation
, params
):
3646 Creates a ns-lcm-opp content to be stored at database.
3647 :param nsr_id: internal id of the instance
3648 :param operation: instantiate, terminate, scale, action, ...
3649 :param params: user parameters for the operation
3650 :return: dictionary following SOL005 format
3652 # Raise exception if invalid arguments
3653 if not (nsr_id
and operation
and params
):
3655 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3662 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3663 "operationState": "PROCESSING",
3664 "statusEnteredTime": now
,
3665 "nsInstanceId": nsr_id
,
3666 "lcmOperationType": operation
,
3668 "isAutomaticInvocation": False,
3669 "operationParams": params
,
3670 "isCancelPending": False,
3672 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3673 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3678 def _format_additional_params(self
, params
):
3679 params
= params
or {}
3680 for key
, value
in params
.items():
3681 if str(value
).startswith("!!yaml "):
3682 params
[key
] = yaml
.safe_load(value
[7:])
3685 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3686 primitive
= seq
.get("name")
3687 primitive_params
= {}
3689 "member_vnf_index": vnf_index
,
3690 "primitive": primitive
,
3691 "primitive_params": primitive_params
,
3694 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3698 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3699 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3700 if op
.get("operationState") == "COMPLETED":
3701 # b. Skip sub-operation
3702 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3703 return self
.SUBOPERATION_STATUS_SKIP
3705 # c. retry executing sub-operation
3706 # The sub-operation exists, and operationState != 'COMPLETED'
3707 # Update operationState = 'PROCESSING' to indicate a retry.
3708 operationState
= "PROCESSING"
3709 detailed_status
= "In progress"
3710 self
._update
_suboperation
_status
(
3711 db_nslcmop
, op_index
, operationState
, detailed_status
3713 # Return the sub-operation index
3714 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3715 # with arguments extracted from the sub-operation
3718 # Find a sub-operation where all keys in a matching dictionary must match
3719 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3720 def _find_suboperation(self
, db_nslcmop
, match
):
3721 if db_nslcmop
and match
:
3722 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3723 for i
, op
in enumerate(op_list
):
3724 if all(op
.get(k
) == match
[k
] for k
in match
):
3726 return self
.SUBOPERATION_STATUS_NOT_FOUND
3728 # Update status for a sub-operation given its index
3729 def _update_suboperation_status(
3730 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3732 # Update DB for HA tasks
3733 q_filter
= {"_id": db_nslcmop
["_id"]}
3735 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3736 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3739 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3742 # Add sub-operation, return the index of the added sub-operation
3743 # Optionally, set operationState, detailed-status, and operationType
3744 # Status and type are currently set for 'scale' sub-operations:
3745 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3746 # 'detailed-status' : status message
3747 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3748 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3749 def _add_suboperation(
3757 mapped_primitive_params
,
3758 operationState
=None,
3759 detailed_status
=None,
3762 RO_scaling_info
=None,
3765 return self
.SUBOPERATION_STATUS_NOT_FOUND
3766 # Get the "_admin.operations" list, if it exists
3767 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3768 op_list
= db_nslcmop_admin
.get("operations")
3769 # Create or append to the "_admin.operations" list
3771 "member_vnf_index": vnf_index
,
3773 "vdu_count_index": vdu_count_index
,
3774 "primitive": primitive
,
3775 "primitive_params": mapped_primitive_params
,
3778 new_op
["operationState"] = operationState
3780 new_op
["detailed-status"] = detailed_status
3782 new_op
["lcmOperationType"] = operationType
3784 new_op
["RO_nsr_id"] = RO_nsr_id
3786 new_op
["RO_scaling_info"] = RO_scaling_info
3788 # No existing operations, create key 'operations' with current operation as first list element
3789 db_nslcmop_admin
.update({"operations": [new_op
]})
3790 op_list
= db_nslcmop_admin
.get("operations")
3792 # Existing operations, append operation to list
3793 op_list
.append(new_op
)
3795 db_nslcmop_update
= {"_admin.operations": op_list
}
3796 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3797 op_index
= len(op_list
) - 1
3800 # Helper methods for scale() sub-operations
3802 # pre-scale/post-scale:
3803 # Check for 3 different cases:
3804 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3805 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3806 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3807 def _check_or_add_scale_suboperation(
3811 vnf_config_primitive
,
3815 RO_scaling_info
=None,
3817 # Find this sub-operation
3818 if RO_nsr_id
and RO_scaling_info
:
3819 operationType
= "SCALE-RO"
3821 "member_vnf_index": vnf_index
,
3822 "RO_nsr_id": RO_nsr_id
,
3823 "RO_scaling_info": RO_scaling_info
,
3827 "member_vnf_index": vnf_index
,
3828 "primitive": vnf_config_primitive
,
3829 "primitive_params": primitive_params
,
3830 "lcmOperationType": operationType
,
3832 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3833 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3834 # a. New sub-operation
3835 # The sub-operation does not exist, add it.
3836 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3837 # The following parameters are set to None for all kind of scaling:
3839 vdu_count_index
= None
3841 if RO_nsr_id
and RO_scaling_info
:
3842 vnf_config_primitive
= None
3843 primitive_params
= None
3846 RO_scaling_info
= None
3847 # Initial status for sub-operation
3848 operationState
= "PROCESSING"
3849 detailed_status
= "In progress"
3850 # Add sub-operation for pre/post-scaling (zero or more operations)
3851 self
._add
_suboperation
(
3857 vnf_config_primitive
,
3865 return self
.SUBOPERATION_STATUS_NEW
3867 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3868 # or op_index (operationState != 'COMPLETED')
3869 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3871 # Function to return execution_environment id
3873 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3874 # TODO vdu_index_count
3875 for vca
in vca_deployed_list
:
3876 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3879 async def destroy_N2VC(
3887 exec_primitives
=True,
3892 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3893 :param logging_text:
3895 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3896 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3897 :param vca_index: index in the database _admin.deployed.VCA
3898 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3899 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3900 not executed properly
3901 :param scaling_in: True destroys the application, False destroys the model
3902 :return: None or exception
3907 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3908 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3912 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3914 # execute terminate_primitives
3916 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3917 config_descriptor
.get("terminate-config-primitive"),
3918 vca_deployed
.get("ee_descriptor_id"),
3920 vdu_id
= vca_deployed
.get("vdu_id")
3921 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3922 vdu_name
= vca_deployed
.get("vdu_name")
3923 vnf_index
= vca_deployed
.get("member-vnf-index")
3924 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3925 for seq
in terminate_primitives
:
3926 # For each sequence in list, get primitive and call _ns_execute_primitive()
3927 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3928 vnf_index
, seq
.get("name")
3930 self
.logger
.debug(logging_text
+ step
)
3931 # Create the primitive for each sequence, i.e. "primitive": "touch"
3932 primitive
= seq
.get("name")
3933 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3938 self
._add
_suboperation
(
3945 mapped_primitive_params
,
3947 # Sub-operations: Call _ns_execute_primitive() instead of action()
3949 result
, result_detail
= await self
._ns
_execute
_primitive
(
3950 vca_deployed
["ee_id"],
3952 mapped_primitive_params
,
3956 except LcmException
:
3957 # this happens when VCA is not deployed. In this case it is not needed to terminate
3959 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3960 if result
not in result_ok
:
3962 "terminate_primitive {} for vnf_member_index={} fails with "
3963 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3965 # set that this VCA do not need terminated
3966 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3970 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3973 # Delete Prometheus Jobs if any
3974 # This uses NSR_ID, so it will destroy any jobs under this index
3975 self
.db
.del_list("prometheus_jobs", {"nsr_id": db_nslcmop
["nsInstanceId"]})
3978 await self
.vca_map
[vca_type
].delete_execution_environment(
3979 vca_deployed
["ee_id"],
3980 scaling_in
=scaling_in
,
3985 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3986 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3987 namespace
= "." + db_nsr
["_id"]
3989 await self
.n2vc
.delete_namespace(
3990 namespace
=namespace
,
3991 total_timeout
=self
.timeout_charm_delete
,
3994 except N2VCNotFound
: # already deleted. Skip
3996 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3998 async def _terminate_RO(
3999 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4002 Terminates a deployment from RO
4003 :param logging_text:
4004 :param nsr_deployed: db_nsr._admin.deployed
4007 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
4008 this method will update only the index 2, but it will write on database the concatenated content of the list
4013 ro_nsr_id
= ro_delete_action
= None
4014 if nsr_deployed
and nsr_deployed
.get("RO"):
4015 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
4016 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
4019 stage
[2] = "Deleting ns from VIM."
4020 db_nsr_update
["detailed-status"] = " ".join(stage
)
4021 self
._write
_op
_status
(nslcmop_id
, stage
)
4022 self
.logger
.debug(logging_text
+ stage
[2])
4023 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4024 self
._write
_op
_status
(nslcmop_id
, stage
)
4025 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
4026 ro_delete_action
= desc
["action_id"]
4028 "_admin.deployed.RO.nsr_delete_action_id"
4029 ] = ro_delete_action
4030 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4031 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4032 if ro_delete_action
:
4033 # wait until NS is deleted from VIM
4034 stage
[2] = "Waiting ns deleted from VIM."
4035 detailed_status_old
= None
4039 + " RO_id={} ro_delete_action={}".format(
4040 ro_nsr_id
, ro_delete_action
4043 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4044 self
._write
_op
_status
(nslcmop_id
, stage
)
4046 delete_timeout
= 20 * 60 # 20 minutes
4047 while delete_timeout
> 0:
4048 desc
= await self
.RO
.show(
4050 item_id_name
=ro_nsr_id
,
4051 extra_item
="action",
4052 extra_item_id
=ro_delete_action
,
4056 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4058 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4059 if ns_status
== "ERROR":
4060 raise ROclient
.ROClientException(ns_status_info
)
4061 elif ns_status
== "BUILD":
4062 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
4063 elif ns_status
== "ACTIVE":
4064 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4065 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4070 ), "ROclient.check_action_status returns unknown {}".format(
4073 if stage
[2] != detailed_status_old
:
4074 detailed_status_old
= stage
[2]
4075 db_nsr_update
["detailed-status"] = " ".join(stage
)
4076 self
._write
_op
_status
(nslcmop_id
, stage
)
4077 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4078 await asyncio
.sleep(5, loop
=self
.loop
)
4080 else: # delete_timeout <= 0:
4081 raise ROclient
.ROClientException(
4082 "Timeout waiting ns deleted from VIM"
4085 except Exception as e
:
4086 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4088 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4090 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4091 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4092 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4094 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
4097 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4099 failed_detail
.append("delete conflict: {}".format(e
))
4102 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
4105 failed_detail
.append("delete error: {}".format(e
))
4107 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
4111 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
4112 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
4114 stage
[2] = "Deleting nsd from RO."
4115 db_nsr_update
["detailed-status"] = " ".join(stage
)
4116 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4117 self
._write
_op
_status
(nslcmop_id
, stage
)
4118 await self
.RO
.delete("nsd", ro_nsd_id
)
4120 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
4122 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4123 except Exception as e
:
4125 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4127 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4129 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
4132 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4134 failed_detail
.append(
4135 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
4137 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4139 failed_detail
.append(
4140 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
4142 self
.logger
.error(logging_text
+ failed_detail
[-1])
4144 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
4145 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
4146 if not vnf_deployed
or not vnf_deployed
["id"]:
4149 ro_vnfd_id
= vnf_deployed
["id"]
4152 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
4153 vnf_deployed
["member-vnf-index"], ro_vnfd_id
4155 db_nsr_update
["detailed-status"] = " ".join(stage
)
4156 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4157 self
._write
_op
_status
(nslcmop_id
, stage
)
4158 await self
.RO
.delete("vnfd", ro_vnfd_id
)
4160 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
4162 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
4163 except Exception as e
:
4165 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4168 "_admin.deployed.RO.vnfd.{}.id".format(index
)
4172 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4175 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4177 failed_detail
.append(
4178 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4180 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4182 failed_detail
.append(
4183 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4185 self
.logger
.error(logging_text
+ failed_detail
[-1])
4188 stage
[2] = "Error deleting from VIM"
4190 stage
[2] = "Deleted from VIM"
4191 db_nsr_update
["detailed-status"] = " ".join(stage
)
4192 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4193 self
._write
_op
_status
(nslcmop_id
, stage
)
4196 raise LcmException("; ".join(failed_detail
))
4198 async def terminate(self
, nsr_id
, nslcmop_id
):
4199 # Try to lock HA task here
4200 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4201 if not task_is_locked_by_me
:
4204 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4205 self
.logger
.debug(logging_text
+ "Enter")
4206 timeout_ns_terminate
= self
.timeout_ns_terminate
4209 operation_params
= None
4211 error_list
= [] # annotates all failed error messages
4212 db_nslcmop_update
= {}
4213 autoremove
= False # autoremove after terminated
4214 tasks_dict_info
= {}
4217 "Stage 1/3: Preparing task.",
4218 "Waiting for previous operations to terminate.",
4221 # ^ contains [stage, step, VIM-status]
4223 # wait for any previous tasks in process
4224 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4226 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4227 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4228 operation_params
= db_nslcmop
.get("operationParams") or {}
4229 if operation_params
.get("timeout_ns_terminate"):
4230 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4231 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4232 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4234 db_nsr_update
["operational-status"] = "terminating"
4235 db_nsr_update
["config-status"] = "terminating"
4236 self
._write
_ns
_status
(
4238 ns_state
="TERMINATING",
4239 current_operation
="TERMINATING",
4240 current_operation_id
=nslcmop_id
,
4241 other_update
=db_nsr_update
,
4243 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4244 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4245 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4248 stage
[1] = "Getting vnf descriptors from db."
4249 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4251 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4253 db_vnfds_from_id
= {}
4254 db_vnfds_from_member_index
= {}
4256 for vnfr
in db_vnfrs_list
:
4257 vnfd_id
= vnfr
["vnfd-id"]
4258 if vnfd_id
not in db_vnfds_from_id
:
4259 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4260 db_vnfds_from_id
[vnfd_id
] = vnfd
4261 db_vnfds_from_member_index
[
4262 vnfr
["member-vnf-index-ref"]
4263 ] = db_vnfds_from_id
[vnfd_id
]
4265 # Destroy individual execution environments when there are terminating primitives.
4266 # Rest of EE will be deleted at once
4267 # TODO - check before calling _destroy_N2VC
4268 # if not operation_params.get("skip_terminate_primitives"):#
4269 # or not vca.get("needed_terminate"):
4270 stage
[0] = "Stage 2/3 execute terminating primitives."
4271 self
.logger
.debug(logging_text
+ stage
[0])
4272 stage
[1] = "Looking execution environment that needs terminate."
4273 self
.logger
.debug(logging_text
+ stage
[1])
4275 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4276 config_descriptor
= None
4277 vca_member_vnf_index
= vca
.get("member-vnf-index")
4278 vca_id
= self
.get_vca_id(
4279 db_vnfrs_dict
.get(vca_member_vnf_index
)
4280 if vca_member_vnf_index
4284 if not vca
or not vca
.get("ee_id"):
4286 if not vca
.get("member-vnf-index"):
4288 config_descriptor
= db_nsr
.get("ns-configuration")
4289 elif vca
.get("vdu_id"):
4290 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4291 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4292 elif vca
.get("kdu_name"):
4293 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4294 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4296 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4297 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4298 vca_type
= vca
.get("type")
4299 exec_terminate_primitives
= not operation_params
.get(
4300 "skip_terminate_primitives"
4301 ) and vca
.get("needed_terminate")
4302 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4303 # pending native charms
4305 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4307 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4308 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4309 task
= asyncio
.ensure_future(
4317 exec_terminate_primitives
,
4321 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4323 # wait for pending tasks of terminate primitives
4327 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4329 error_list
= await self
._wait
_for
_tasks
(
4332 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4336 tasks_dict_info
.clear()
4338 return # raise LcmException("; ".join(error_list))
4340 # remove All execution environments at once
4341 stage
[0] = "Stage 3/3 delete all."
4343 if nsr_deployed
.get("VCA"):
4344 stage
[1] = "Deleting all execution environments."
4345 self
.logger
.debug(logging_text
+ stage
[1])
4346 vca_id
= self
.get_vca_id({}, db_nsr
)
4347 task_delete_ee
= asyncio
.ensure_future(
4349 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4350 timeout
=self
.timeout_charm_delete
,
4353 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4354 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4356 # Delete from k8scluster
4357 stage
[1] = "Deleting KDUs."
4358 self
.logger
.debug(logging_text
+ stage
[1])
4359 # print(nsr_deployed)
4360 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4361 if not kdu
or not kdu
.get("kdu-instance"):
4363 kdu_instance
= kdu
.get("kdu-instance")
4364 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4365 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4366 vca_id
= self
.get_vca_id({}, db_nsr
)
4367 task_delete_kdu_instance
= asyncio
.ensure_future(
4368 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4369 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4370 kdu_instance
=kdu_instance
,
4377 + "Unknown k8s deployment type {}".format(
4378 kdu
.get("k8scluster-type")
4383 task_delete_kdu_instance
4384 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4387 stage
[1] = "Deleting ns from VIM."
4389 task_delete_ro
= asyncio
.ensure_future(
4390 self
._terminate
_ng
_ro
(
4391 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4395 task_delete_ro
= asyncio
.ensure_future(
4397 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4400 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4402 # rest of staff will be done at finally
4405 ROclient
.ROClientException
,
4410 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4412 except asyncio
.CancelledError
:
4414 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4416 exc
= "Operation was cancelled"
4417 except Exception as e
:
4418 exc
= traceback
.format_exc()
4419 self
.logger
.critical(
4420 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4425 error_list
.append(str(exc
))
4427 # wait for pending tasks
4429 stage
[1] = "Waiting for terminate pending tasks."
4430 self
.logger
.debug(logging_text
+ stage
[1])
4431 error_list
+= await self
._wait
_for
_tasks
(
4434 timeout_ns_terminate
,
4438 stage
[1] = stage
[2] = ""
4439 except asyncio
.CancelledError
:
4440 error_list
.append("Cancelled")
4441 # TODO cancell all tasks
4442 except Exception as exc
:
4443 error_list
.append(str(exc
))
4444 # update status at database
4446 error_detail
= "; ".join(error_list
)
4447 # self.logger.error(logging_text + error_detail)
4448 error_description_nslcmop
= "{} Detail: {}".format(
4449 stage
[0], error_detail
4451 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4452 nslcmop_id
, stage
[0]
4455 db_nsr_update
["operational-status"] = "failed"
4456 db_nsr_update
["detailed-status"] = (
4457 error_description_nsr
+ " Detail: " + error_detail
4459 db_nslcmop_update
["detailed-status"] = error_detail
4460 nslcmop_operation_state
= "FAILED"
4464 error_description_nsr
= error_description_nslcmop
= None
4465 ns_state
= "NOT_INSTANTIATED"
4466 db_nsr_update
["operational-status"] = "terminated"
4467 db_nsr_update
["detailed-status"] = "Done"
4468 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4469 db_nslcmop_update
["detailed-status"] = "Done"
4470 nslcmop_operation_state
= "COMPLETED"
4473 self
._write
_ns
_status
(
4476 current_operation
="IDLE",
4477 current_operation_id
=None,
4478 error_description
=error_description_nsr
,
4479 error_detail
=error_detail
,
4480 other_update
=db_nsr_update
,
4482 self
._write
_op
_status
(
4485 error_message
=error_description_nslcmop
,
4486 operation_state
=nslcmop_operation_state
,
4487 other_update
=db_nslcmop_update
,
4489 if ns_state
== "NOT_INSTANTIATED":
4493 {"nsr-id-ref": nsr_id
},
4494 {"_admin.nsState": "NOT_INSTANTIATED"},
4496 except DbException
as e
:
4499 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4503 if operation_params
:
4504 autoremove
= operation_params
.get("autoremove", False)
4505 if nslcmop_operation_state
:
4507 await self
.msg
.aiowrite(
4512 "nslcmop_id": nslcmop_id
,
4513 "operationState": nslcmop_operation_state
,
4514 "autoremove": autoremove
,
4518 except Exception as e
:
4520 logging_text
+ "kafka_write notification Exception {}".format(e
)
4523 self
.logger
.debug(logging_text
+ "Exit")
4524 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4526 async def _wait_for_tasks(
4527 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4530 error_detail_list
= []
4532 pending_tasks
= list(created_tasks_info
.keys())
4533 num_tasks
= len(pending_tasks
)
4535 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4536 self
._write
_op
_status
(nslcmop_id
, stage
)
4537 while pending_tasks
:
4539 _timeout
= timeout
+ time_start
- time()
4540 done
, pending_tasks
= await asyncio
.wait(
4541 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4543 num_done
+= len(done
)
4544 if not done
: # Timeout
4545 for task
in pending_tasks
:
4546 new_error
= created_tasks_info
[task
] + ": Timeout"
4547 error_detail_list
.append(new_error
)
4548 error_list
.append(new_error
)
4551 if task
.cancelled():
4554 exc
= task
.exception()
4556 if isinstance(exc
, asyncio
.TimeoutError
):
4558 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4559 error_list
.append(created_tasks_info
[task
])
4560 error_detail_list
.append(new_error
)
4567 ROclient
.ROClientException
,
4573 self
.logger
.error(logging_text
+ new_error
)
4575 exc_traceback
= "".join(
4576 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4580 + created_tasks_info
[task
]
4586 logging_text
+ created_tasks_info
[task
] + ": Done"
4588 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4590 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4591 if nsr_id
: # update also nsr
4596 "errorDescription": "Error at: " + ", ".join(error_list
),
4597 "errorDetail": ". ".join(error_detail_list
),
4600 self
._write
_op
_status
(nslcmop_id
, stage
)
4601 return error_detail_list
4604 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4606 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4607 The default-value is used. If it is between < > it look for a value at instantiation_params
4608 :param primitive_desc: portion of VNFD/NSD that describes primitive
4609 :param params: Params provided by user
4610 :param instantiation_params: Instantiation params provided by user
4611 :return: a dictionary with the calculated params
4613 calculated_params
= {}
4614 for parameter
in primitive_desc
.get("parameter", ()):
4615 param_name
= parameter
["name"]
4616 if param_name
in params
:
4617 calculated_params
[param_name
] = params
[param_name
]
4618 elif "default-value" in parameter
or "value" in parameter
:
4619 if "value" in parameter
:
4620 calculated_params
[param_name
] = parameter
["value"]
4622 calculated_params
[param_name
] = parameter
["default-value"]
4624 isinstance(calculated_params
[param_name
], str)
4625 and calculated_params
[param_name
].startswith("<")
4626 and calculated_params
[param_name
].endswith(">")
4628 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4629 calculated_params
[param_name
] = instantiation_params
[
4630 calculated_params
[param_name
][1:-1]
4634 "Parameter {} needed to execute primitive {} not provided".format(
4635 calculated_params
[param_name
], primitive_desc
["name"]
4640 "Parameter {} needed to execute primitive {} not provided".format(
4641 param_name
, primitive_desc
["name"]
4645 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4646 calculated_params
[param_name
] = yaml
.safe_dump(
4647 calculated_params
[param_name
], default_flow_style
=True, width
=256
4649 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4651 ].startswith("!!yaml "):
4652 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4653 if parameter
.get("data-type") == "INTEGER":
4655 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4656 except ValueError: # error converting string to int
4658 "Parameter {} of primitive {} must be integer".format(
4659 param_name
, primitive_desc
["name"]
4662 elif parameter
.get("data-type") == "BOOLEAN":
4663 calculated_params
[param_name
] = not (
4664 (str(calculated_params
[param_name
])).lower() == "false"
4667 # add always ns_config_info if primitive name is config
4668 if primitive_desc
["name"] == "config":
4669 if "ns_config_info" in instantiation_params
:
4670 calculated_params
["ns_config_info"] = instantiation_params
[
4673 return calculated_params
4675 def _look_for_deployed_vca(
4682 ee_descriptor_id
=None,
4684 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4685 for vca
in deployed_vca
:
4688 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4691 vdu_count_index
is not None
4692 and vdu_count_index
!= vca
["vdu_count_index"]
4695 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4697 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4701 # vca_deployed not found
4703 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4704 " is not deployed".format(
4713 ee_id
= vca
.get("ee_id")
4715 "type", "lxc_proxy_charm"
4716 ) # default value for backward compatibility - proxy charm
4719 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4720 "execution environment".format(
4721 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4724 return ee_id
, vca_type
4726 async def _ns_execute_primitive(
4732 retries_interval
=30,
4739 if primitive
== "config":
4740 primitive_params
= {"params": primitive_params
}
4742 vca_type
= vca_type
or "lxc_proxy_charm"
4746 output
= await asyncio
.wait_for(
4747 self
.vca_map
[vca_type
].exec_primitive(
4749 primitive_name
=primitive
,
4750 params_dict
=primitive_params
,
4751 progress_timeout
=self
.timeout_progress_primitive
,
4752 total_timeout
=self
.timeout_primitive
,
4757 timeout
=timeout
or self
.timeout_primitive
,
4761 except asyncio
.CancelledError
:
4763 except Exception as e
: # asyncio.TimeoutError
4764 if isinstance(e
, asyncio
.TimeoutError
):
4769 "Error executing action {} on {} -> {}".format(
4774 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4776 return "FAILED", str(e
)
4778 return "COMPLETED", output
4780 except (LcmException
, asyncio
.CancelledError
):
4782 except Exception as e
:
4783 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4785 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4787 Updating the vca_status with latest juju information in nsrs record
4788 :param: nsr_id: Id of the nsr
4789 :param: nslcmop_id: Id of the nslcmop
4793 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4794 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4795 vca_id
= self
.get_vca_id({}, db_nsr
)
4796 if db_nsr
["_admin"]["deployed"]["K8s"]:
4797 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4798 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4799 await self
._on
_update
_k
8s
_db
(
4800 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4803 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4804 table
, filter = "nsrs", {"_id": nsr_id
}
4805 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4806 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4808 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4809 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4811 async def action(self
, nsr_id
, nslcmop_id
):
4812 # Try to lock HA task here
4813 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4814 if not task_is_locked_by_me
:
4817 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4818 self
.logger
.debug(logging_text
+ "Enter")
4819 # get all needed from database
4823 db_nslcmop_update
= {}
4824 nslcmop_operation_state
= None
4825 error_description_nslcmop
= None
4828 # wait for any previous tasks in process
4829 step
= "Waiting for previous operations to terminate"
4830 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4832 self
._write
_ns
_status
(
4835 current_operation
="RUNNING ACTION",
4836 current_operation_id
=nslcmop_id
,
4839 step
= "Getting information from database"
4840 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4841 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4842 if db_nslcmop
["operationParams"].get("primitive_params"):
4843 db_nslcmop
["operationParams"]["primitive_params"] = json
.loads(
4844 db_nslcmop
["operationParams"]["primitive_params"]
4847 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4848 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4849 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4850 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4851 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4852 primitive
= db_nslcmop
["operationParams"]["primitive"]
4853 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4854 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4855 "timeout_ns_action", self
.timeout_primitive
4859 step
= "Getting vnfr from database"
4860 db_vnfr
= self
.db
.get_one(
4861 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4863 step
= "Getting vnfd from database"
4864 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4866 # Sync filesystem before running a primitive
4867 self
.fs
.sync(db_vnfr
["vnfd-id"])
4869 step
= "Getting nsd from database"
4870 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4872 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4873 # for backward compatibility
4874 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4875 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4876 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4877 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4879 # look for primitive
4880 config_primitive_desc
= descriptor_configuration
= None
4882 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4884 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4886 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4888 descriptor_configuration
= db_nsd
.get("ns-configuration")
4890 if descriptor_configuration
and descriptor_configuration
.get(
4893 for config_primitive
in descriptor_configuration
["config-primitive"]:
4894 if config_primitive
["name"] == primitive
:
4895 config_primitive_desc
= config_primitive
4898 if not config_primitive_desc
:
4899 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4901 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4905 primitive_name
= primitive
4906 ee_descriptor_id
= None
4908 primitive_name
= config_primitive_desc
.get(
4909 "execution-environment-primitive", primitive
4911 ee_descriptor_id
= config_primitive_desc
.get(
4912 "execution-environment-ref"
4918 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4920 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4923 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4925 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4927 desc_params
= parse_yaml_strings(
4928 db_vnfr
.get("additionalParamsForVnf")
4931 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4932 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4933 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4935 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4936 actions
.add(primitive
["name"])
4937 for primitive
in kdu_configuration
.get("config-primitive", []):
4938 actions
.add(primitive
["name"])
4939 kdu_action
= True if primitive_name
in actions
else False
4941 # TODO check if ns is in a proper status
4943 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4945 # kdur and desc_params already set from before
4946 if primitive_params
:
4947 desc_params
.update(primitive_params
)
4948 # TODO Check if we will need something at vnf level
4949 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4951 kdu_name
== kdu
["kdu-name"]
4952 and kdu
["member-vnf-index"] == vnf_index
4957 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4960 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4961 msg
= "unknown k8scluster-type '{}'".format(
4962 kdu
.get("k8scluster-type")
4964 raise LcmException(msg
)
4967 "collection": "nsrs",
4968 "filter": {"_id": nsr_id
},
4969 "path": "_admin.deployed.K8s.{}".format(index
),
4973 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4975 step
= "Executing kdu {}".format(primitive_name
)
4976 if primitive_name
== "upgrade":
4977 if desc_params
.get("kdu_model"):
4978 kdu_model
= desc_params
.get("kdu_model")
4979 del desc_params
["kdu_model"]
4981 kdu_model
= kdu
.get("kdu-model")
4982 parts
= kdu_model
.split(sep
=":")
4984 kdu_model
= parts
[0]
4986 detailed_status
= await asyncio
.wait_for(
4987 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4988 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4989 kdu_instance
=kdu
.get("kdu-instance"),
4991 kdu_model
=kdu_model
,
4994 timeout
=timeout_ns_action
,
4996 timeout
=timeout_ns_action
+ 10,
4999 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
5001 elif primitive_name
== "rollback":
5002 detailed_status
= await asyncio
.wait_for(
5003 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
5004 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5005 kdu_instance
=kdu
.get("kdu-instance"),
5008 timeout
=timeout_ns_action
,
5010 elif primitive_name
== "status":
5011 detailed_status
= await asyncio
.wait_for(
5012 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
5013 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5014 kdu_instance
=kdu
.get("kdu-instance"),
5017 timeout
=timeout_ns_action
,
5020 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
5021 kdu
["kdu-name"], nsr_id
5023 params
= self
._map
_primitive
_params
(
5024 config_primitive_desc
, primitive_params
, desc_params
5027 detailed_status
= await asyncio
.wait_for(
5028 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
5029 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5030 kdu_instance
=kdu_instance
,
5031 primitive_name
=primitive_name
,
5034 timeout
=timeout_ns_action
,
5037 timeout
=timeout_ns_action
,
5041 nslcmop_operation_state
= "COMPLETED"
5043 detailed_status
= ""
5044 nslcmop_operation_state
= "FAILED"
5046 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5047 nsr_deployed
["VCA"],
5048 member_vnf_index
=vnf_index
,
5050 vdu_count_index
=vdu_count_index
,
5051 ee_descriptor_id
=ee_descriptor_id
,
5053 for vca_index
, vca_deployed
in enumerate(
5054 db_nsr
["_admin"]["deployed"]["VCA"]
5056 if vca_deployed
.get("member-vnf-index") == vnf_index
:
5058 "collection": "nsrs",
5059 "filter": {"_id": nsr_id
},
5060 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
5064 nslcmop_operation_state
,
5066 ) = await self
._ns
_execute
_primitive
(
5068 primitive
=primitive_name
,
5069 primitive_params
=self
._map
_primitive
_params
(
5070 config_primitive_desc
, primitive_params
, desc_params
5072 timeout
=timeout_ns_action
,
5078 db_nslcmop_update
["detailed-status"] = detailed_status
5079 error_description_nslcmop
= (
5080 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
5084 + " task Done with result {} {}".format(
5085 nslcmop_operation_state
, detailed_status
5088 return # database update is called inside finally
5090 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5091 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5093 except asyncio
.CancelledError
:
5095 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5097 exc
= "Operation was cancelled"
5098 except asyncio
.TimeoutError
:
5099 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5101 except Exception as e
:
5102 exc
= traceback
.format_exc()
5103 self
.logger
.critical(
5104 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5113 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5114 nslcmop_operation_state
= "FAILED"
5116 self
._write
_ns
_status
(
5120 ], # TODO check if degraded. For the moment use previous status
5121 current_operation
="IDLE",
5122 current_operation_id
=None,
5123 # error_description=error_description_nsr,
5124 # error_detail=error_detail,
5125 other_update
=db_nsr_update
,
5128 self
._write
_op
_status
(
5131 error_message
=error_description_nslcmop
,
5132 operation_state
=nslcmop_operation_state
,
5133 other_update
=db_nslcmop_update
,
5136 if nslcmop_operation_state
:
5138 await self
.msg
.aiowrite(
5143 "nslcmop_id": nslcmop_id
,
5144 "operationState": nslcmop_operation_state
,
5148 except Exception as e
:
5150 logging_text
+ "kafka_write notification Exception {}".format(e
)
5152 self
.logger
.debug(logging_text
+ "Exit")
5153 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5154 return nslcmop_operation_state
, detailed_status
5156 async def scale(self
, nsr_id
, nslcmop_id
):
5157 # Try to lock HA task here
5158 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5159 if not task_is_locked_by_me
:
5162 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5163 stage
= ["", "", ""]
5164 tasks_dict_info
= {}
5165 # ^ stage, step, VIM progress
5166 self
.logger
.debug(logging_text
+ "Enter")
5167 # get all needed from database
5169 db_nslcmop_update
= {}
5172 # in case of error, indicates what part of scale was failed to put nsr at error status
5173 scale_process
= None
5174 old_operational_status
= ""
5175 old_config_status
= ""
5178 # wait for any previous tasks in process
5179 step
= "Waiting for previous operations to terminate"
5180 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5181 self
._write
_ns
_status
(
5184 current_operation
="SCALING",
5185 current_operation_id
=nslcmop_id
,
5188 step
= "Getting nslcmop from database"
5190 step
+ " after having waited for previous tasks to be completed"
5192 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5194 step
= "Getting nsr from database"
5195 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5196 old_operational_status
= db_nsr
["operational-status"]
5197 old_config_status
= db_nsr
["config-status"]
5199 step
= "Parsing scaling parameters"
5200 db_nsr_update
["operational-status"] = "scaling"
5201 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5202 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5204 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5206 ]["member-vnf-index"]
5207 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5209 ]["scaling-group-descriptor"]
5210 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5211 # for backward compatibility
5212 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5213 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5214 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5215 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5217 step
= "Getting vnfr from database"
5218 db_vnfr
= self
.db
.get_one(
5219 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5222 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5224 step
= "Getting vnfd from database"
5225 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5227 base_folder
= db_vnfd
["_admin"]["storage"]
5229 step
= "Getting scaling-group-descriptor"
5230 scaling_descriptor
= find_in_list(
5231 get_scaling_aspect(db_vnfd
),
5232 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5234 if not scaling_descriptor
:
5236 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5237 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5240 step
= "Sending scale order to VIM"
5241 # TODO check if ns is in a proper status
5243 if not db_nsr
["_admin"].get("scaling-group"):
5248 "_admin.scaling-group": [
5249 {"name": scaling_group
, "nb-scale-op": 0}
5253 admin_scale_index
= 0
5255 for admin_scale_index
, admin_scale_info
in enumerate(
5256 db_nsr
["_admin"]["scaling-group"]
5258 if admin_scale_info
["name"] == scaling_group
:
5259 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5261 else: # not found, set index one plus last element and add new entry with the name
5262 admin_scale_index
+= 1
5264 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5267 vca_scaling_info
= []
5268 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5269 if scaling_type
== "SCALE_OUT":
5270 if "aspect-delta-details" not in scaling_descriptor
:
5272 "Aspect delta details not fount in scaling descriptor {}".format(
5273 scaling_descriptor
["name"]
5276 # count if max-instance-count is reached
5277 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5279 scaling_info
["scaling_direction"] = "OUT"
5280 scaling_info
["vdu-create"] = {}
5281 scaling_info
["kdu-create"] = {}
5282 for delta
in deltas
:
5283 for vdu_delta
in delta
.get("vdu-delta", {}):
5284 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5285 # vdu_index also provides the number of instance of the targeted vdu
5286 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5287 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5291 additional_params
= (
5292 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5295 cloud_init_list
= []
5297 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5298 max_instance_count
= 10
5299 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5300 max_instance_count
= vdu_profile
.get(
5301 "max-number-of-instances", 10
5304 default_instance_num
= get_number_of_instances(
5307 instances_number
= vdu_delta
.get("number-of-instances", 1)
5308 nb_scale_op
+= instances_number
5310 new_instance_count
= nb_scale_op
+ default_instance_num
5311 # Control if new count is over max and vdu count is less than max.
5312 # Then assign new instance count
5313 if new_instance_count
> max_instance_count
> vdu_count
:
5314 instances_number
= new_instance_count
- max_instance_count
5316 instances_number
= instances_number
5318 if new_instance_count
> max_instance_count
:
5320 "reached the limit of {} (max-instance-count) "
5321 "scaling-out operations for the "
5322 "scaling-group-descriptor '{}'".format(
5323 nb_scale_op
, scaling_group
5326 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5328 # TODO Information of its own ip is not available because db_vnfr is not updated.
5329 additional_params
["OSM"] = get_osm_params(
5330 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5332 cloud_init_list
.append(
5333 self
._parse
_cloud
_init
(
5340 vca_scaling_info
.append(
5342 "osm_vdu_id": vdu_delta
["id"],
5343 "member-vnf-index": vnf_index
,
5345 "vdu_index": vdu_index
+ x
,
5348 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5349 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5350 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5351 kdu_name
= kdu_profile
["kdu-name"]
5352 resource_name
= kdu_profile
.get("resource-name", "")
5354 # Might have different kdus in the same delta
5355 # Should have list for each kdu
5356 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5357 scaling_info
["kdu-create"][kdu_name
] = []
5359 kdur
= get_kdur(db_vnfr
, kdu_name
)
5360 if kdur
.get("helm-chart"):
5361 k8s_cluster_type
= "helm-chart-v3"
5362 self
.logger
.debug("kdur: {}".format(kdur
))
5364 kdur
.get("helm-version")
5365 and kdur
.get("helm-version") == "v2"
5367 k8s_cluster_type
= "helm-chart"
5368 elif kdur
.get("juju-bundle"):
5369 k8s_cluster_type
= "juju-bundle"
5372 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5373 "juju-bundle. Maybe an old NBI version is running".format(
5374 db_vnfr
["member-vnf-index-ref"], kdu_name
5378 max_instance_count
= 10
5379 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5380 max_instance_count
= kdu_profile
.get(
5381 "max-number-of-instances", 10
5384 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5385 deployed_kdu
, _
= get_deployed_kdu(
5386 nsr_deployed
, kdu_name
, vnf_index
5388 if deployed_kdu
is None:
5390 "KDU '{}' for vnf '{}' not deployed".format(
5394 kdu_instance
= deployed_kdu
.get("kdu-instance")
5395 instance_num
= await self
.k8scluster_map
[
5401 cluster_uuid
=deployed_kdu
.get("k8scluster-uuid"),
5402 kdu_model
=deployed_kdu
.get("kdu-model"),
5404 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5405 "number-of-instances", 1
5408 # Control if new count is over max and instance_num is less than max.
5409 # Then assign max instance number to kdu replica count
5410 if kdu_replica_count
> max_instance_count
> instance_num
:
5411 kdu_replica_count
= max_instance_count
5412 if kdu_replica_count
> max_instance_count
:
5414 "reached the limit of {} (max-instance-count) "
5415 "scaling-out operations for the "
5416 "scaling-group-descriptor '{}'".format(
5417 instance_num
, scaling_group
5421 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5422 vca_scaling_info
.append(
5424 "osm_kdu_id": kdu_name
,
5425 "member-vnf-index": vnf_index
,
5427 "kdu_index": instance_num
+ x
- 1,
5430 scaling_info
["kdu-create"][kdu_name
].append(
5432 "member-vnf-index": vnf_index
,
5434 "k8s-cluster-type": k8s_cluster_type
,
5435 "resource-name": resource_name
,
5436 "scale": kdu_replica_count
,
5439 elif scaling_type
== "SCALE_IN":
5440 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5442 scaling_info
["scaling_direction"] = "IN"
5443 scaling_info
["vdu-delete"] = {}
5444 scaling_info
["kdu-delete"] = {}
5446 for delta
in deltas
:
5447 for vdu_delta
in delta
.get("vdu-delta", {}):
5448 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5449 min_instance_count
= 0
5450 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5451 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5452 min_instance_count
= vdu_profile
["min-number-of-instances"]
5454 default_instance_num
= get_number_of_instances(
5455 db_vnfd
, vdu_delta
["id"]
5457 instance_num
= vdu_delta
.get("number-of-instances", 1)
5458 nb_scale_op
-= instance_num
5460 new_instance_count
= nb_scale_op
+ default_instance_num
5462 if new_instance_count
< min_instance_count
< vdu_count
:
5463 instances_number
= min_instance_count
- new_instance_count
5465 instances_number
= instance_num
5467 if new_instance_count
< min_instance_count
:
5469 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5470 "scaling-group-descriptor '{}'".format(
5471 nb_scale_op
, scaling_group
5474 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5475 vca_scaling_info
.append(
5477 "osm_vdu_id": vdu_delta
["id"],
5478 "member-vnf-index": vnf_index
,
5480 "vdu_index": vdu_index
- 1 - x
,
5483 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5484 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5485 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5486 kdu_name
= kdu_profile
["kdu-name"]
5487 resource_name
= kdu_profile
.get("resource-name", "")
5489 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5490 scaling_info
["kdu-delete"][kdu_name
] = []
5492 kdur
= get_kdur(db_vnfr
, kdu_name
)
5493 if kdur
.get("helm-chart"):
5494 k8s_cluster_type
= "helm-chart-v3"
5495 self
.logger
.debug("kdur: {}".format(kdur
))
5497 kdur
.get("helm-version")
5498 and kdur
.get("helm-version") == "v2"
5500 k8s_cluster_type
= "helm-chart"
5501 elif kdur
.get("juju-bundle"):
5502 k8s_cluster_type
= "juju-bundle"
5505 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5506 "juju-bundle. Maybe an old NBI version is running".format(
5507 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5511 min_instance_count
= 0
5512 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5513 min_instance_count
= kdu_profile
["min-number-of-instances"]
5515 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5516 deployed_kdu
, _
= get_deployed_kdu(
5517 nsr_deployed
, kdu_name
, vnf_index
5519 if deployed_kdu
is None:
5521 "KDU '{}' for vnf '{}' not deployed".format(
5525 kdu_instance
= deployed_kdu
.get("kdu-instance")
5526 instance_num
= await self
.k8scluster_map
[
5532 cluster_uuid
=deployed_kdu
.get("k8scluster-uuid"),
5533 kdu_model
=deployed_kdu
.get("kdu-model"),
5535 kdu_replica_count
= instance_num
- kdu_delta
.get(
5536 "number-of-instances", 1
5539 if kdu_replica_count
< min_instance_count
< instance_num
:
5540 kdu_replica_count
= min_instance_count
5541 if kdu_replica_count
< min_instance_count
:
5543 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5544 "scaling-group-descriptor '{}'".format(
5545 instance_num
, scaling_group
5549 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5550 vca_scaling_info
.append(
5552 "osm_kdu_id": kdu_name
,
5553 "member-vnf-index": vnf_index
,
5555 "kdu_index": instance_num
- x
- 1,
5558 scaling_info
["kdu-delete"][kdu_name
].append(
5560 "member-vnf-index": vnf_index
,
5562 "k8s-cluster-type": k8s_cluster_type
,
5563 "resource-name": resource_name
,
5564 "scale": kdu_replica_count
,
5568 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5569 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5570 if scaling_info
["scaling_direction"] == "IN":
5571 for vdur
in reversed(db_vnfr
["vdur"]):
5572 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5573 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5574 scaling_info
["vdu"].append(
5576 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5577 "vdu_id": vdur
["vdu-id-ref"],
5581 for interface
in vdur
["interfaces"]:
5582 scaling_info
["vdu"][-1]["interface"].append(
5584 "name": interface
["name"],
5585 "ip_address": interface
["ip-address"],
5586 "mac_address": interface
.get("mac-address"),
5589 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5592 step
= "Executing pre-scale vnf-config-primitive"
5593 if scaling_descriptor
.get("scaling-config-action"):
5594 for scaling_config_action
in scaling_descriptor
[
5595 "scaling-config-action"
5598 scaling_config_action
.get("trigger") == "pre-scale-in"
5599 and scaling_type
== "SCALE_IN"
5601 scaling_config_action
.get("trigger") == "pre-scale-out"
5602 and scaling_type
== "SCALE_OUT"
5604 vnf_config_primitive
= scaling_config_action
[
5605 "vnf-config-primitive-name-ref"
5607 step
= db_nslcmop_update
[
5609 ] = "executing pre-scale scaling-config-action '{}'".format(
5610 vnf_config_primitive
5613 # look for primitive
5614 for config_primitive
in (
5615 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5616 ).get("config-primitive", ()):
5617 if config_primitive
["name"] == vnf_config_primitive
:
5621 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5622 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5623 "primitive".format(scaling_group
, vnf_config_primitive
)
5626 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5627 if db_vnfr
.get("additionalParamsForVnf"):
5628 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5630 scale_process
= "VCA"
5631 db_nsr_update
["config-status"] = "configuring pre-scaling"
5632 primitive_params
= self
._map
_primitive
_params
(
5633 config_primitive
, {}, vnfr_params
5636 # Pre-scale retry check: Check if this sub-operation has been executed before
5637 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5640 vnf_config_primitive
,
5644 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5645 # Skip sub-operation
5646 result
= "COMPLETED"
5647 result_detail
= "Done"
5650 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5651 vnf_config_primitive
, result
, result_detail
5655 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5656 # New sub-operation: Get index of this sub-operation
5658 len(db_nslcmop
.get("_admin", {}).get("operations"))
5663 + "vnf_config_primitive={} New sub-operation".format(
5664 vnf_config_primitive
5668 # retry: Get registered params for this existing sub-operation
5669 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5672 vnf_index
= op
.get("member_vnf_index")
5673 vnf_config_primitive
= op
.get("primitive")
5674 primitive_params
= op
.get("primitive_params")
5677 + "vnf_config_primitive={} Sub-operation retry".format(
5678 vnf_config_primitive
5681 # Execute the primitive, either with new (first-time) or registered (reintent) args
5682 ee_descriptor_id
= config_primitive
.get(
5683 "execution-environment-ref"
5685 primitive_name
= config_primitive
.get(
5686 "execution-environment-primitive", vnf_config_primitive
5688 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5689 nsr_deployed
["VCA"],
5690 member_vnf_index
=vnf_index
,
5692 vdu_count_index
=None,
5693 ee_descriptor_id
=ee_descriptor_id
,
5695 result
, result_detail
= await self
._ns
_execute
_primitive
(
5704 + "vnf_config_primitive={} Done with result {} {}".format(
5705 vnf_config_primitive
, result
, result_detail
5708 # Update operationState = COMPLETED | FAILED
5709 self
._update
_suboperation
_status
(
5710 db_nslcmop
, op_index
, result
, result_detail
5713 if result
== "FAILED":
5714 raise LcmException(result_detail
)
5715 db_nsr_update
["config-status"] = old_config_status
5716 scale_process
= None
5720 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5723 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5726 # SCALE-IN VCA - BEGIN
5727 if vca_scaling_info
:
5728 step
= db_nslcmop_update
[
5730 ] = "Deleting the execution environments"
5731 scale_process
= "VCA"
5732 for vca_info
in vca_scaling_info
:
5733 if vca_info
["type"] == "delete":
5734 member_vnf_index
= str(vca_info
["member-vnf-index"])
5736 logging_text
+ "vdu info: {}".format(vca_info
)
5738 if vca_info
.get("osm_vdu_id"):
5739 vdu_id
= vca_info
["osm_vdu_id"]
5740 vdu_index
= int(vca_info
["vdu_index"])
5743 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5744 member_vnf_index
, vdu_id
, vdu_index
5748 kdu_id
= vca_info
["osm_kdu_id"]
5751 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5752 member_vnf_index
, kdu_id
, vdu_index
5754 stage
[2] = step
= "Scaling in VCA"
5755 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5756 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5757 config_update
= db_nsr
["configurationStatus"]
5758 for vca_index
, vca
in enumerate(vca_update
):
5760 (vca
or vca
.get("ee_id"))
5761 and vca
["member-vnf-index"] == member_vnf_index
5762 and vca
["vdu_count_index"] == vdu_index
5764 if vca
.get("vdu_id"):
5765 config_descriptor
= get_configuration(
5766 db_vnfd
, vca
.get("vdu_id")
5768 elif vca
.get("kdu_name"):
5769 config_descriptor
= get_configuration(
5770 db_vnfd
, vca
.get("kdu_name")
5773 config_descriptor
= get_configuration(
5774 db_vnfd
, db_vnfd
["id"]
5776 operation_params
= (
5777 db_nslcmop
.get("operationParams") or {}
5779 exec_terminate_primitives
= not operation_params
.get(
5780 "skip_terminate_primitives"
5781 ) and vca
.get("needed_terminate")
5782 task
= asyncio
.ensure_future(
5791 exec_primitives
=exec_terminate_primitives
,
5795 timeout
=self
.timeout_charm_delete
,
5798 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5801 del vca_update
[vca_index
]
5802 del config_update
[vca_index
]
5803 # wait for pending tasks of terminate primitives
5807 + "Waiting for tasks {}".format(
5808 list(tasks_dict_info
.keys())
5811 error_list
= await self
._wait
_for
_tasks
(
5815 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5820 tasks_dict_info
.clear()
5822 raise LcmException("; ".join(error_list
))
5824 db_vca_and_config_update
= {
5825 "_admin.deployed.VCA": vca_update
,
5826 "configurationStatus": config_update
,
5829 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5831 scale_process
= None
5832 # SCALE-IN VCA - END
5835 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5836 scale_process
= "RO"
5837 if self
.ro_config
.get("ng"):
5838 await self
._scale
_ng
_ro
(
5839 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5841 scaling_info
.pop("vdu-create", None)
5842 scaling_info
.pop("vdu-delete", None)
5844 scale_process
= None
5848 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5849 scale_process
= "KDU"
5850 await self
._scale
_kdu
(
5851 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5853 scaling_info
.pop("kdu-create", None)
5854 scaling_info
.pop("kdu-delete", None)
5856 scale_process
= None
5860 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5862 # SCALE-UP VCA - BEGIN
5863 if vca_scaling_info
:
5864 step
= db_nslcmop_update
[
5866 ] = "Creating new execution environments"
5867 scale_process
= "VCA"
5868 for vca_info
in vca_scaling_info
:
5869 if vca_info
["type"] == "create":
5870 member_vnf_index
= str(vca_info
["member-vnf-index"])
5872 logging_text
+ "vdu info: {}".format(vca_info
)
5874 vnfd_id
= db_vnfr
["vnfd-ref"]
5875 if vca_info
.get("osm_vdu_id"):
5876 vdu_index
= int(vca_info
["vdu_index"])
5877 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5878 if db_vnfr
.get("additionalParamsForVnf"):
5879 deploy_params
.update(
5881 db_vnfr
["additionalParamsForVnf"].copy()
5884 descriptor_config
= get_configuration(
5885 db_vnfd
, db_vnfd
["id"]
5887 if descriptor_config
:
5892 logging_text
=logging_text
5893 + "member_vnf_index={} ".format(member_vnf_index
),
5896 nslcmop_id
=nslcmop_id
,
5902 member_vnf_index
=member_vnf_index
,
5903 vdu_index
=vdu_index
,
5905 deploy_params
=deploy_params
,
5906 descriptor_config
=descriptor_config
,
5907 base_folder
=base_folder
,
5908 task_instantiation_info
=tasks_dict_info
,
5911 vdu_id
= vca_info
["osm_vdu_id"]
5912 vdur
= find_in_list(
5913 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5915 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5916 if vdur
.get("additionalParams"):
5917 deploy_params_vdu
= parse_yaml_strings(
5918 vdur
["additionalParams"]
5921 deploy_params_vdu
= deploy_params
5922 deploy_params_vdu
["OSM"] = get_osm_params(
5923 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5925 if descriptor_config
:
5930 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5931 member_vnf_index
, vdu_id
, vdu_index
5933 stage
[2] = step
= "Scaling out VCA"
5934 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5936 logging_text
=logging_text
5937 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5938 member_vnf_index
, vdu_id
, vdu_index
5942 nslcmop_id
=nslcmop_id
,
5948 member_vnf_index
=member_vnf_index
,
5949 vdu_index
=vdu_index
,
5951 deploy_params
=deploy_params_vdu
,
5952 descriptor_config
=descriptor_config
,
5953 base_folder
=base_folder
,
5954 task_instantiation_info
=tasks_dict_info
,
5958 kdu_name
= vca_info
["osm_kdu_id"]
5959 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5960 if descriptor_config
:
5962 kdu_index
= int(vca_info
["kdu_index"])
5966 for x
in db_vnfr
["kdur"]
5967 if x
["kdu-name"] == kdu_name
5969 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5970 if kdur
.get("additionalParams"):
5971 deploy_params_kdu
= parse_yaml_strings(
5972 kdur
["additionalParams"]
5976 logging_text
=logging_text
,
5979 nslcmop_id
=nslcmop_id
,
5985 member_vnf_index
=member_vnf_index
,
5986 vdu_index
=kdu_index
,
5988 deploy_params
=deploy_params_kdu
,
5989 descriptor_config
=descriptor_config
,
5990 base_folder
=base_folder
,
5991 task_instantiation_info
=tasks_dict_info
,
5994 # SCALE-UP VCA - END
5995 scale_process
= None
5998 # execute primitive service POST-SCALING
5999 step
= "Executing post-scale vnf-config-primitive"
6000 if scaling_descriptor
.get("scaling-config-action"):
6001 for scaling_config_action
in scaling_descriptor
[
6002 "scaling-config-action"
6005 scaling_config_action
.get("trigger") == "post-scale-in"
6006 and scaling_type
== "SCALE_IN"
6008 scaling_config_action
.get("trigger") == "post-scale-out"
6009 and scaling_type
== "SCALE_OUT"
6011 vnf_config_primitive
= scaling_config_action
[
6012 "vnf-config-primitive-name-ref"
6014 step
= db_nslcmop_update
[
6016 ] = "executing post-scale scaling-config-action '{}'".format(
6017 vnf_config_primitive
6020 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
6021 if db_vnfr
.get("additionalParamsForVnf"):
6022 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
6024 # look for primitive
6025 for config_primitive
in (
6026 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
6027 ).get("config-primitive", ()):
6028 if config_primitive
["name"] == vnf_config_primitive
:
6032 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
6033 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
6034 "config-primitive".format(
6035 scaling_group
, vnf_config_primitive
6038 scale_process
= "VCA"
6039 db_nsr_update
["config-status"] = "configuring post-scaling"
6040 primitive_params
= self
._map
_primitive
_params
(
6041 config_primitive
, {}, vnfr_params
6044 # Post-scale retry check: Check if this sub-operation has been executed before
6045 op_index
= self
._check
_or
_add
_scale
_suboperation
(
6048 vnf_config_primitive
,
6052 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
6053 # Skip sub-operation
6054 result
= "COMPLETED"
6055 result_detail
= "Done"
6058 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
6059 vnf_config_primitive
, result
, result_detail
6063 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
6064 # New sub-operation: Get index of this sub-operation
6066 len(db_nslcmop
.get("_admin", {}).get("operations"))
6071 + "vnf_config_primitive={} New sub-operation".format(
6072 vnf_config_primitive
6076 # retry: Get registered params for this existing sub-operation
6077 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
6080 vnf_index
= op
.get("member_vnf_index")
6081 vnf_config_primitive
= op
.get("primitive")
6082 primitive_params
= op
.get("primitive_params")
6085 + "vnf_config_primitive={} Sub-operation retry".format(
6086 vnf_config_primitive
6089 # Execute the primitive, either with new (first-time) or registered (reintent) args
6090 ee_descriptor_id
= config_primitive
.get(
6091 "execution-environment-ref"
6093 primitive_name
= config_primitive
.get(
6094 "execution-environment-primitive", vnf_config_primitive
6096 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6097 nsr_deployed
["VCA"],
6098 member_vnf_index
=vnf_index
,
6100 vdu_count_index
=None,
6101 ee_descriptor_id
=ee_descriptor_id
,
6103 result
, result_detail
= await self
._ns
_execute
_primitive
(
6112 + "vnf_config_primitive={} Done with result {} {}".format(
6113 vnf_config_primitive
, result
, result_detail
6116 # Update operationState = COMPLETED | FAILED
6117 self
._update
_suboperation
_status
(
6118 db_nslcmop
, op_index
, result
, result_detail
6121 if result
== "FAILED":
6122 raise LcmException(result_detail
)
6123 db_nsr_update
["config-status"] = old_config_status
6124 scale_process
= None
6129 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6130 db_nsr_update
["operational-status"] = (
6132 if old_operational_status
== "failed"
6133 else old_operational_status
6135 db_nsr_update
["config-status"] = old_config_status
6138 ROclient
.ROClientException
,
6143 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6145 except asyncio
.CancelledError
:
6147 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6149 exc
= "Operation was cancelled"
6150 except Exception as e
:
6151 exc
= traceback
.format_exc()
6152 self
.logger
.critical(
6153 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6157 self
._write
_ns
_status
(
6160 current_operation
="IDLE",
6161 current_operation_id
=None,
6164 stage
[1] = "Waiting for instantiate pending tasks."
6165 self
.logger
.debug(logging_text
+ stage
[1])
6166 exc
= await self
._wait
_for
_tasks
(
6169 self
.timeout_ns_deploy
,
6177 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6178 nslcmop_operation_state
= "FAILED"
6180 db_nsr_update
["operational-status"] = old_operational_status
6181 db_nsr_update
["config-status"] = old_config_status
6182 db_nsr_update
["detailed-status"] = ""
6184 if "VCA" in scale_process
:
6185 db_nsr_update
["config-status"] = "failed"
6186 if "RO" in scale_process
:
6187 db_nsr_update
["operational-status"] = "failed"
6190 ] = "FAILED scaling nslcmop={} {}: {}".format(
6191 nslcmop_id
, step
, exc
6194 error_description_nslcmop
= None
6195 nslcmop_operation_state
= "COMPLETED"
6196 db_nslcmop_update
["detailed-status"] = "Done"
6198 self
._write
_op
_status
(
6201 error_message
=error_description_nslcmop
,
6202 operation_state
=nslcmop_operation_state
,
6203 other_update
=db_nslcmop_update
,
6206 self
._write
_ns
_status
(
6209 current_operation
="IDLE",
6210 current_operation_id
=None,
6211 other_update
=db_nsr_update
,
6214 if nslcmop_operation_state
:
6218 "nslcmop_id": nslcmop_id
,
6219 "operationState": nslcmop_operation_state
,
6221 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6222 except Exception as e
:
6224 logging_text
+ "kafka_write notification Exception {}".format(e
)
6226 self
.logger
.debug(logging_text
+ "Exit")
6227 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6229 async def _scale_kdu(
6230 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6232 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6233 for kdu_name
in _scaling_info
:
6234 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6235 deployed_kdu
, index
= get_deployed_kdu(
6236 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6238 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6239 kdu_instance
= deployed_kdu
["kdu-instance"]
6240 kdu_model
= deployed_kdu
.get("kdu-model")
6241 scale
= int(kdu_scaling_info
["scale"])
6242 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6245 "collection": "nsrs",
6246 "filter": {"_id": nsr_id
},
6247 "path": "_admin.deployed.K8s.{}".format(index
),
6250 step
= "scaling application {}".format(
6251 kdu_scaling_info
["resource-name"]
6253 self
.logger
.debug(logging_text
+ step
)
6255 if kdu_scaling_info
["type"] == "delete":
6256 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6259 and kdu_config
.get("terminate-config-primitive")
6260 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6262 terminate_config_primitive_list
= kdu_config
.get(
6263 "terminate-config-primitive"
6265 terminate_config_primitive_list
.sort(
6266 key
=lambda val
: int(val
["seq"])
6270 terminate_config_primitive
6271 ) in terminate_config_primitive_list
:
6272 primitive_params_
= self
._map
_primitive
_params
(
6273 terminate_config_primitive
, {}, {}
6275 step
= "execute terminate config primitive"
6276 self
.logger
.debug(logging_text
+ step
)
6277 await asyncio
.wait_for(
6278 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6279 cluster_uuid
=cluster_uuid
,
6280 kdu_instance
=kdu_instance
,
6281 primitive_name
=terminate_config_primitive
["name"],
6282 params
=primitive_params_
,
6289 await asyncio
.wait_for(
6290 self
.k8scluster_map
[k8s_cluster_type
].scale(
6293 kdu_scaling_info
["resource-name"],
6295 cluster_uuid
=cluster_uuid
,
6296 kdu_model
=kdu_model
,
6300 timeout
=self
.timeout_vca_on_error
,
6303 if kdu_scaling_info
["type"] == "create":
6304 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6307 and kdu_config
.get("initial-config-primitive")
6308 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6310 initial_config_primitive_list
= kdu_config
.get(
6311 "initial-config-primitive"
6313 initial_config_primitive_list
.sort(
6314 key
=lambda val
: int(val
["seq"])
6317 for initial_config_primitive
in initial_config_primitive_list
:
6318 primitive_params_
= self
._map
_primitive
_params
(
6319 initial_config_primitive
, {}, {}
6321 step
= "execute initial config primitive"
6322 self
.logger
.debug(logging_text
+ step
)
6323 await asyncio
.wait_for(
6324 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6325 cluster_uuid
=cluster_uuid
,
6326 kdu_instance
=kdu_instance
,
6327 primitive_name
=initial_config_primitive
["name"],
6328 params
=primitive_params_
,
6335 async def _scale_ng_ro(
6336 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6338 nsr_id
= db_nslcmop
["nsInstanceId"]
6339 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6342 # read from db: vnfd's for every vnf
6345 # for each vnf in ns, read vnfd
6346 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6347 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6348 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6349 # if we haven't this vnfd, read it from db
6350 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6352 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6353 db_vnfds
.append(vnfd
)
6354 n2vc_key
= self
.n2vc
.get_public_key()
6355 n2vc_key_list
= [n2vc_key
]
6358 vdu_scaling_info
.get("vdu-create"),
6359 vdu_scaling_info
.get("vdu-delete"),
6362 # db_vnfr has been updated, update db_vnfrs to use it
6363 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6364 await self
._instantiate
_ng
_ro
(
6374 start_deploy
=time(),
6375 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6377 if vdu_scaling_info
.get("vdu-delete"):
6379 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6382 async def extract_prometheus_scrape_jobs(
6386 ee_config_descriptor
,
6391 # look if exist a file called 'prometheus*.j2' and
6392 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6396 for f
in artifact_content
6397 if f
.startswith("prometheus") and f
.endswith(".j2")
6403 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6407 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6408 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6410 vnfr_id
= vnfr_id
.replace("-", "")
6412 "JOB_NAME": vnfr_id
,
6413 "TARGET_IP": target_ip
,
6414 "EXPORTER_POD_IP": host_name
,
6415 "EXPORTER_POD_PORT": host_port
,
6417 job_list
= parse_job(job_data
, variables
)
6418 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6419 for job
in job_list
:
6421 not isinstance(job
.get("job_name"), str)
6422 or vnfr_id
not in job
["job_name"]
6424 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6425 job
["nsr_id"] = nsr_id
6426 job
["vnfr_id"] = vnfr_id
6429 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6431 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6433 :param: vim_account_id: VIM Account ID
6435 :return: (cloud_name, cloud_credential)
6437 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6438 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6440 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6442 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6444 :param: vim_account_id: VIM Account ID
6446 :return: (cloud_name, cloud_credential)
6448 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6449 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")