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 ns_params
= db_nslcmop
.get("operationParams")
2280 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2281 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2283 timeout_ns_deploy
= self
.timeout
.get(
2284 "ns_deploy", self
.timeout_ns_deploy
2288 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2289 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2290 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2291 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2292 self
.fs
.sync(db_nsr
["nsd-id"])
2294 # nsr_name = db_nsr["name"] # TODO short-name??
2296 # read from db: vnf's of this ns
2297 stage
[1] = "Getting vnfrs from db."
2298 self
.logger
.debug(logging_text
+ stage
[1])
2299 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2301 # read from db: vnfd's for every vnf
2302 db_vnfds
= [] # every vnfd data
2304 # for each vnf in ns, read vnfd
2305 for vnfr
in db_vnfrs_list
:
2306 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2307 vnfd_id
= vnfr
["vnfd-id"]
2308 vnfd_ref
= vnfr
["vnfd-ref"]
2309 self
.fs
.sync(vnfd_id
)
2311 # if we haven't this vnfd, read it from db
2312 if vnfd_id
not in db_vnfds
:
2314 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2317 self
.logger
.debug(logging_text
+ stage
[1])
2318 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2321 db_vnfds
.append(vnfd
)
2323 # Get or generates the _admin.deployed.VCA list
2324 vca_deployed_list
= None
2325 if db_nsr
["_admin"].get("deployed"):
2326 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2327 if vca_deployed_list
is None:
2328 vca_deployed_list
= []
2329 configuration_status_list
= []
2330 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2331 db_nsr_update
["configurationStatus"] = configuration_status_list
2332 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2333 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2334 elif isinstance(vca_deployed_list
, dict):
2335 # maintain backward compatibility. Change a dict to list at database
2336 vca_deployed_list
= list(vca_deployed_list
.values())
2337 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2338 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2341 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2343 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2344 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2346 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2347 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2348 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2350 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2353 # n2vc_redesign STEP 2 Deploy Network Scenario
2354 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2355 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2357 stage
[1] = "Deploying KDUs."
2358 # self.logger.debug(logging_text + "Before deploy_kdus")
2359 # Call to deploy_kdus in case exists the "vdu:kdu" param
2360 await self
.deploy_kdus(
2361 logging_text
=logging_text
,
2363 nslcmop_id
=nslcmop_id
,
2366 task_instantiation_info
=tasks_dict_info
,
2369 stage
[1] = "Getting VCA public key."
2370 # n2vc_redesign STEP 1 Get VCA public ssh-key
2371 # feature 1429. Add n2vc public key to needed VMs
2372 n2vc_key
= self
.n2vc
.get_public_key()
2373 n2vc_key_list
= [n2vc_key
]
2374 if self
.vca_config
.get("public_key"):
2375 n2vc_key_list
.append(self
.vca_config
["public_key"])
2377 stage
[1] = "Deploying NS at VIM."
2378 task_ro
= asyncio
.ensure_future(
2379 self
.instantiate_RO(
2380 logging_text
=logging_text
,
2384 db_nslcmop
=db_nslcmop
,
2387 n2vc_key_list
=n2vc_key_list
,
2391 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2392 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2394 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2395 stage
[1] = "Deploying Execution Environments."
2396 self
.logger
.debug(logging_text
+ stage
[1])
2398 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2399 for vnf_profile
in get_vnf_profiles(nsd
):
2400 vnfd_id
= vnf_profile
["vnfd-id"]
2401 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2402 member_vnf_index
= str(vnf_profile
["id"])
2403 db_vnfr
= db_vnfrs
[member_vnf_index
]
2404 base_folder
= vnfd
["_admin"]["storage"]
2410 # Get additional parameters
2411 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2412 if db_vnfr
.get("additionalParamsForVnf"):
2413 deploy_params
.update(
2414 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2417 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2418 if descriptor_config
:
2420 logging_text
=logging_text
2421 + "member_vnf_index={} ".format(member_vnf_index
),
2424 nslcmop_id
=nslcmop_id
,
2430 member_vnf_index
=member_vnf_index
,
2431 vdu_index
=vdu_index
,
2433 deploy_params
=deploy_params
,
2434 descriptor_config
=descriptor_config
,
2435 base_folder
=base_folder
,
2436 task_instantiation_info
=tasks_dict_info
,
2440 # Deploy charms for each VDU that supports one.
2441 for vdud
in get_vdu_list(vnfd
):
2443 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2444 vdur
= find_in_list(
2445 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2448 if vdur
.get("additionalParams"):
2449 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2451 deploy_params_vdu
= deploy_params
2452 deploy_params_vdu
["OSM"] = get_osm_params(
2453 db_vnfr
, vdu_id
, vdu_count_index
=0
2455 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2457 self
.logger
.debug("VDUD > {}".format(vdud
))
2459 "Descriptor config > {}".format(descriptor_config
)
2461 if descriptor_config
:
2464 for vdu_index
in range(vdud_count
):
2465 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2467 logging_text
=logging_text
2468 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2469 member_vnf_index
, vdu_id
, vdu_index
2473 nslcmop_id
=nslcmop_id
,
2479 member_vnf_index
=member_vnf_index
,
2480 vdu_index
=vdu_index
,
2482 deploy_params
=deploy_params_vdu
,
2483 descriptor_config
=descriptor_config
,
2484 base_folder
=base_folder
,
2485 task_instantiation_info
=tasks_dict_info
,
2488 for kdud
in get_kdu_list(vnfd
):
2489 kdu_name
= kdud
["name"]
2490 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2491 if descriptor_config
:
2496 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2498 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2499 if kdur
.get("additionalParams"):
2500 deploy_params_kdu
= parse_yaml_strings(
2501 kdur
["additionalParams"]
2505 logging_text
=logging_text
,
2508 nslcmop_id
=nslcmop_id
,
2514 member_vnf_index
=member_vnf_index
,
2515 vdu_index
=vdu_index
,
2517 deploy_params
=deploy_params_kdu
,
2518 descriptor_config
=descriptor_config
,
2519 base_folder
=base_folder
,
2520 task_instantiation_info
=tasks_dict_info
,
2524 # Check if this NS has a charm configuration
2525 descriptor_config
= nsd
.get("ns-configuration")
2526 if descriptor_config
and descriptor_config
.get("juju"):
2529 member_vnf_index
= None
2535 # Get additional parameters
2536 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2537 if db_nsr
.get("additionalParamsForNs"):
2538 deploy_params
.update(
2539 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2541 base_folder
= nsd
["_admin"]["storage"]
2543 logging_text
=logging_text
,
2546 nslcmop_id
=nslcmop_id
,
2552 member_vnf_index
=member_vnf_index
,
2553 vdu_index
=vdu_index
,
2555 deploy_params
=deploy_params
,
2556 descriptor_config
=descriptor_config
,
2557 base_folder
=base_folder
,
2558 task_instantiation_info
=tasks_dict_info
,
2562 # rest of staff will be done at finally
2565 ROclient
.ROClientException
,
2571 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2574 except asyncio
.CancelledError
:
2576 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2578 exc
= "Operation was cancelled"
2579 except Exception as e
:
2580 exc
= traceback
.format_exc()
2581 self
.logger
.critical(
2582 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2587 error_list
.append(str(exc
))
2589 # wait for pending tasks
2591 stage
[1] = "Waiting for instantiate pending tasks."
2592 self
.logger
.debug(logging_text
+ stage
[1])
2593 error_list
+= await self
._wait
_for
_tasks
(
2601 stage
[1] = stage
[2] = ""
2602 except asyncio
.CancelledError
:
2603 error_list
.append("Cancelled")
2604 # TODO cancel all tasks
2605 except Exception as exc
:
2606 error_list
.append(str(exc
))
2608 # update operation-status
2609 db_nsr_update
["operational-status"] = "running"
2610 # let's begin with VCA 'configured' status (later we can change it)
2611 db_nsr_update
["config-status"] = "configured"
2612 for task
, task_name
in tasks_dict_info
.items():
2613 if not task
.done() or task
.cancelled() or task
.exception():
2614 if task_name
.startswith(self
.task_name_deploy_vca
):
2615 # A N2VC task is pending
2616 db_nsr_update
["config-status"] = "failed"
2618 # RO or KDU task is pending
2619 db_nsr_update
["operational-status"] = "failed"
2621 # update status at database
2623 error_detail
= ". ".join(error_list
)
2624 self
.logger
.error(logging_text
+ error_detail
)
2625 error_description_nslcmop
= "{} Detail: {}".format(
2626 stage
[0], error_detail
2628 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2629 nslcmop_id
, stage
[0]
2632 db_nsr_update
["detailed-status"] = (
2633 error_description_nsr
+ " Detail: " + error_detail
2635 db_nslcmop_update
["detailed-status"] = error_detail
2636 nslcmop_operation_state
= "FAILED"
2640 error_description_nsr
= error_description_nslcmop
= None
2642 db_nsr_update
["detailed-status"] = "Done"
2643 db_nslcmop_update
["detailed-status"] = "Done"
2644 nslcmop_operation_state
= "COMPLETED"
2647 self
._write
_ns
_status
(
2650 current_operation
="IDLE",
2651 current_operation_id
=None,
2652 error_description
=error_description_nsr
,
2653 error_detail
=error_detail
,
2654 other_update
=db_nsr_update
,
2656 self
._write
_op
_status
(
2659 error_message
=error_description_nslcmop
,
2660 operation_state
=nslcmop_operation_state
,
2661 other_update
=db_nslcmop_update
,
2664 if nslcmop_operation_state
:
2666 await self
.msg
.aiowrite(
2671 "nslcmop_id": nslcmop_id
,
2672 "operationState": nslcmop_operation_state
,
2676 except Exception as e
:
2678 logging_text
+ "kafka_write notification Exception {}".format(e
)
2681 self
.logger
.debug(logging_text
+ "Exit")
2682 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2684 def _get_vnfd(self
, vnfd_id
: str, cached_vnfds
: Dict
[str, Any
]):
2685 if vnfd_id
not in cached_vnfds
:
2686 cached_vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
2687 return cached_vnfds
[vnfd_id
]
2689 def _get_vnfr(self
, nsr_id
: str, vnf_profile_id
: str, cached_vnfrs
: Dict
[str, Any
]):
2690 if vnf_profile_id
not in cached_vnfrs
:
2691 cached_vnfrs
[vnf_profile_id
] = self
.db
.get_one(
2694 "member-vnf-index-ref": vnf_profile_id
,
2695 "nsr-id-ref": nsr_id
,
2698 return cached_vnfrs
[vnf_profile_id
]
2700 def _is_deployed_vca_in_relation(
2701 self
, vca
: DeployedVCA
, relation
: Relation
2704 for endpoint
in (relation
.provider
, relation
.requirer
):
2705 if endpoint
["kdu-resource-profile-id"]:
2708 vca
.vnf_profile_id
== endpoint
.vnf_profile_id
2709 and vca
.vdu_profile_id
== endpoint
.vdu_profile_id
2710 and vca
.execution_environment_ref
== endpoint
.execution_environment_ref
2716 def _update_ee_relation_data_with_implicit_data(
2717 self
, nsr_id
, nsd
, ee_relation_data
, cached_vnfds
, vnf_profile_id
: str = None
2719 ee_relation_data
= safe_get_ee_relation(
2720 nsr_id
, ee_relation_data
, vnf_profile_id
=vnf_profile_id
2722 ee_relation_level
= EELevel
.get_level(ee_relation_data
)
2723 if (ee_relation_level
in (EELevel
.VNF
, EELevel
.VDU
)) and not ee_relation_data
[
2724 "execution-environment-ref"
2726 vnf_profile
= get_vnf_profile(nsd
, ee_relation_data
["vnf-profile-id"])
2727 vnfd_id
= vnf_profile
["vnfd-id"]
2728 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2731 if ee_relation_level
== EELevel
.VNF
2732 else ee_relation_data
["vdu-profile-id"]
2734 ee
= get_juju_ee_ref(db_vnfd
, entity_id
)
2737 f
"not execution environments found for ee_relation {ee_relation_data}"
2739 ee_relation_data
["execution-environment-ref"] = ee
["id"]
2740 return ee_relation_data
2742 def _get_ns_relations(
2745 nsd
: Dict
[str, Any
],
2747 cached_vnfds
: Dict
[str, Any
],
2748 ) -> List
[Relation
]:
2750 db_ns_relations
= get_ns_configuration_relation_list(nsd
)
2751 for r
in db_ns_relations
:
2752 provider_dict
= None
2753 requirer_dict
= None
2754 if all(key
in r
for key
in ("provider", "requirer")):
2755 provider_dict
= r
["provider"]
2756 requirer_dict
= r
["requirer"]
2757 elif "entities" in r
:
2758 provider_id
= r
["entities"][0]["id"]
2761 "endpoint": r
["entities"][0]["endpoint"],
2763 if provider_id
!= nsd
["id"]:
2764 provider_dict
["vnf-profile-id"] = provider_id
2765 requirer_id
= r
["entities"][1]["id"]
2768 "endpoint": r
["entities"][1]["endpoint"],
2770 if requirer_id
!= nsd
["id"]:
2771 requirer_dict
["vnf-profile-id"] = requirer_id
2773 raise Exception("provider/requirer or entities must be included in the relation.")
2774 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2775 nsr_id
, nsd
, provider_dict
, cached_vnfds
2777 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2778 nsr_id
, nsd
, requirer_dict
, cached_vnfds
2780 provider
= EERelation(relation_provider
)
2781 requirer
= EERelation(relation_requirer
)
2782 relation
= Relation(r
["name"], provider
, requirer
)
2783 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2785 relations
.append(relation
)
2788 def _get_vnf_relations(
2791 nsd
: Dict
[str, Any
],
2793 cached_vnfds
: Dict
[str, Any
],
2794 ) -> List
[Relation
]:
2796 vnf_profile
= get_vnf_profile(nsd
, vca
.vnf_profile_id
)
2797 vnf_profile_id
= vnf_profile
["id"]
2798 vnfd_id
= vnf_profile
["vnfd-id"]
2799 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2800 db_vnf_relations
= get_relation_list(db_vnfd
, vnfd_id
)
2801 for r
in db_vnf_relations
:
2802 provider_dict
= None
2803 requirer_dict
= None
2804 if all(key
in r
for key
in ("provider", "requirer")):
2805 provider_dict
= r
["provider"]
2806 requirer_dict
= r
["requirer"]
2807 elif "entities" in r
:
2808 provider_id
= r
["entities"][0]["id"]
2811 "vnf-profile-id": vnf_profile_id
,
2812 "endpoint": r
["entities"][0]["endpoint"],
2814 if provider_id
!= vnfd_id
:
2815 provider_dict
["vdu-profile-id"] = provider_id
2816 requirer_id
= r
["entities"][1]["id"]
2819 "vnf-profile-id": vnf_profile_id
,
2820 "endpoint": r
["entities"][1]["endpoint"],
2822 if requirer_id
!= vnfd_id
:
2823 requirer_dict
["vdu-profile-id"] = requirer_id
2825 raise Exception("provider/requirer or entities must be included in the relation.")
2826 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2827 nsr_id
, nsd
, provider_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2829 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2830 nsr_id
, nsd
, requirer_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2832 provider
= EERelation(relation_provider
)
2833 requirer
= EERelation(relation_requirer
)
2834 relation
= Relation(r
["name"], provider
, requirer
)
2835 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2837 relations
.append(relation
)
2840 def _get_kdu_resource_data(
2842 ee_relation
: EERelation
,
2843 db_nsr
: Dict
[str, Any
],
2844 cached_vnfds
: Dict
[str, Any
],
2845 ) -> DeployedK8sResource
:
2846 nsd
= get_nsd(db_nsr
)
2847 vnf_profiles
= get_vnf_profiles(nsd
)
2848 vnfd_id
= find_in_list(
2850 lambda vnf_profile
: vnf_profile
["id"] == ee_relation
.vnf_profile_id
,
2852 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2853 kdu_resource_profile
= get_kdu_resource_profile(
2854 db_vnfd
, ee_relation
.kdu_resource_profile_id
2856 kdu_name
= kdu_resource_profile
["kdu-name"]
2857 deployed_kdu
, _
= get_deployed_kdu(
2858 db_nsr
.get("_admin", ()).get("deployed", ()),
2860 ee_relation
.vnf_profile_id
,
2862 deployed_kdu
.update({"resource-name": kdu_resource_profile
["resource-name"]})
2865 def _get_deployed_component(
2867 ee_relation
: EERelation
,
2868 db_nsr
: Dict
[str, Any
],
2869 cached_vnfds
: Dict
[str, Any
],
2870 ) -> DeployedComponent
:
2871 nsr_id
= db_nsr
["_id"]
2872 deployed_component
= None
2873 ee_level
= EELevel
.get_level(ee_relation
)
2874 if ee_level
== EELevel
.NS
:
2875 vca
= get_deployed_vca(db_nsr
, {"vdu_id": None, "member-vnf-index": None})
2877 deployed_component
= DeployedVCA(nsr_id
, vca
)
2878 elif ee_level
== EELevel
.VNF
:
2879 vca
= get_deployed_vca(
2883 "member-vnf-index": ee_relation
.vnf_profile_id
,
2884 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2888 deployed_component
= DeployedVCA(nsr_id
, vca
)
2889 elif ee_level
== EELevel
.VDU
:
2890 vca
= get_deployed_vca(
2893 "vdu_id": ee_relation
.vdu_profile_id
,
2894 "member-vnf-index": ee_relation
.vnf_profile_id
,
2895 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2899 deployed_component
= DeployedVCA(nsr_id
, vca
)
2900 elif ee_level
== EELevel
.KDU
:
2901 kdu_resource_data
= self
._get
_kdu
_resource
_data
(
2902 ee_relation
, db_nsr
, cached_vnfds
2904 if kdu_resource_data
:
2905 deployed_component
= DeployedK8sResource(kdu_resource_data
)
2906 return deployed_component
2908 async def _add_relation(
2912 db_nsr
: Dict
[str, Any
],
2913 cached_vnfds
: Dict
[str, Any
],
2914 cached_vnfrs
: Dict
[str, Any
],
2916 deployed_provider
= self
._get
_deployed
_component
(
2917 relation
.provider
, db_nsr
, cached_vnfds
2919 deployed_requirer
= self
._get
_deployed
_component
(
2920 relation
.requirer
, db_nsr
, cached_vnfds
2924 and deployed_requirer
2925 and deployed_provider
.config_sw_installed
2926 and deployed_requirer
.config_sw_installed
2928 provider_db_vnfr
= (
2930 relation
.provider
.nsr_id
,
2931 relation
.provider
.vnf_profile_id
,
2934 if relation
.provider
.vnf_profile_id
2937 requirer_db_vnfr
= (
2939 relation
.requirer
.nsr_id
,
2940 relation
.requirer
.vnf_profile_id
,
2943 if relation
.requirer
.vnf_profile_id
2946 provider_vca_id
= self
.get_vca_id(provider_db_vnfr
, db_nsr
)
2947 requirer_vca_id
= self
.get_vca_id(requirer_db_vnfr
, db_nsr
)
2948 provider_relation_endpoint
= RelationEndpoint(
2949 deployed_provider
.ee_id
,
2951 relation
.provider
.endpoint
,
2953 requirer_relation_endpoint
= RelationEndpoint(
2954 deployed_requirer
.ee_id
,
2956 relation
.requirer
.endpoint
,
2958 await self
.vca_map
[vca_type
].add_relation(
2959 provider
=provider_relation_endpoint
,
2960 requirer
=requirer_relation_endpoint
,
2962 # remove entry from relations list
2966 async def _add_vca_relations(
2972 timeout
: int = 3600,
2976 # 1. find all relations for this VCA
2977 # 2. wait for other peers related
2981 # STEP 1: find all relations for this VCA
2984 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2985 nsd
= get_nsd(db_nsr
)
2988 deployed_vca_dict
= get_deployed_vca_list(db_nsr
)[vca_index
]
2989 my_vca
= DeployedVCA(nsr_id
, deployed_vca_dict
)
2994 relations
.extend(self
._get
_ns
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
2995 relations
.extend(self
._get
_vnf
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
2997 # if no relations, terminate
2999 self
.logger
.debug(logging_text
+ " No relations")
3002 self
.logger
.debug(logging_text
+ " adding relations {}".format(relations
))
3009 if now
- start
>= timeout
:
3010 self
.logger
.error(logging_text
+ " : timeout adding relations")
3013 # reload nsr from database (we need to update record: _admin.deployed.VCA)
3014 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3016 # for each relation, find the VCA's related
3017 for relation
in relations
.copy():
3018 added
= await self
._add
_relation
(
3026 relations
.remove(relation
)
3029 self
.logger
.debug("Relations added")
3031 await asyncio
.sleep(5.0)
3035 except Exception as e
:
3036 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
3039 async def _install_kdu(
3047 k8s_instance_info
: dict,
3048 k8params
: dict = None,
3054 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
3057 "collection": "nsrs",
3058 "filter": {"_id": nsr_id
},
3059 "path": nsr_db_path
,
3062 if k8s_instance_info
.get("kdu-deployment-name"):
3063 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
3065 kdu_instance
= self
.k8scluster_map
[
3067 ].generate_kdu_instance_name(
3068 db_dict
=db_dict_install
,
3069 kdu_model
=k8s_instance_info
["kdu-model"],
3070 kdu_name
=k8s_instance_info
["kdu-name"],
3073 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3075 await self
.k8scluster_map
[k8sclustertype
].install(
3076 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3077 kdu_model
=k8s_instance_info
["kdu-model"],
3080 db_dict
=db_dict_install
,
3082 kdu_name
=k8s_instance_info
["kdu-name"],
3083 namespace
=k8s_instance_info
["namespace"],
3084 kdu_instance
=kdu_instance
,
3088 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3091 # Obtain services to obtain management service ip
3092 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
3093 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3094 kdu_instance
=kdu_instance
,
3095 namespace
=k8s_instance_info
["namespace"],
3098 # Obtain management service info (if exists)
3099 vnfr_update_dict
= {}
3100 kdu_config
= get_configuration(vnfd
, kdud
["name"])
3102 target_ee_list
= kdu_config
.get("execution-environment-list", [])
3107 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
3110 for service
in kdud
.get("service", [])
3111 if service
.get("mgmt-service")
3113 for mgmt_service
in mgmt_services
:
3114 for service
in services
:
3115 if service
["name"].startswith(mgmt_service
["name"]):
3116 # Mgmt service found, Obtain service ip
3117 ip
= service
.get("external_ip", service
.get("cluster_ip"))
3118 if isinstance(ip
, list) and len(ip
) == 1:
3122 "kdur.{}.ip-address".format(kdu_index
)
3125 # Check if must update also mgmt ip at the vnf
3126 service_external_cp
= mgmt_service
.get(
3127 "external-connection-point-ref"
3129 if service_external_cp
:
3131 deep_get(vnfd
, ("mgmt-interface", "cp"))
3132 == service_external_cp
3134 vnfr_update_dict
["ip-address"] = ip
3139 "external-connection-point-ref", ""
3141 == service_external_cp
,
3144 "kdur.{}.ip-address".format(kdu_index
)
3149 "Mgmt service name: {} not found".format(
3150 mgmt_service
["name"]
3154 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3155 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3157 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3160 and kdu_config
.get("initial-config-primitive")
3161 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3163 initial_config_primitive_list
= kdu_config
.get(
3164 "initial-config-primitive"
3166 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3168 for initial_config_primitive
in initial_config_primitive_list
:
3169 primitive_params_
= self
._map
_primitive
_params
(
3170 initial_config_primitive
, {}, {}
3173 await asyncio
.wait_for(
3174 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3175 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3176 kdu_instance
=kdu_instance
,
3177 primitive_name
=initial_config_primitive
["name"],
3178 params
=primitive_params_
,
3179 db_dict
=db_dict_install
,
3185 except Exception as e
:
3186 # Prepare update db with error and raise exception
3189 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3193 vnfr_data
.get("_id"),
3194 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3197 # ignore to keep original exception
3199 # reraise original error
3204 async def deploy_kdus(
3211 task_instantiation_info
,
3213 # Launch kdus if present in the descriptor
3215 k8scluster_id_2_uuic
= {
3216 "helm-chart-v3": {},
3221 async def _get_cluster_id(cluster_id
, cluster_type
):
3222 nonlocal k8scluster_id_2_uuic
3223 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3224 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3226 # check if K8scluster is creating and wait look if previous tasks in process
3227 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3228 "k8scluster", cluster_id
3231 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3232 task_name
, cluster_id
3234 self
.logger
.debug(logging_text
+ text
)
3235 await asyncio
.wait(task_dependency
, timeout
=3600)
3237 db_k8scluster
= self
.db
.get_one(
3238 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3240 if not db_k8scluster
:
3241 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3243 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3245 if cluster_type
== "helm-chart-v3":
3247 # backward compatibility for existing clusters that have not been initialized for helm v3
3248 k8s_credentials
= yaml
.safe_dump(
3249 db_k8scluster
.get("credentials")
3251 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3252 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3254 db_k8scluster_update
= {}
3255 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3256 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3257 db_k8scluster_update
[
3258 "_admin.helm-chart-v3.created"
3260 db_k8scluster_update
[
3261 "_admin.helm-chart-v3.operationalState"
3264 "k8sclusters", cluster_id
, db_k8scluster_update
3266 except Exception as e
:
3269 + "error initializing helm-v3 cluster: {}".format(str(e
))
3272 "K8s cluster '{}' has not been initialized for '{}'".format(
3273 cluster_id
, cluster_type
3278 "K8s cluster '{}' has not been initialized for '{}'".format(
3279 cluster_id
, cluster_type
3282 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3285 logging_text
+= "Deploy kdus: "
3288 db_nsr_update
= {"_admin.deployed.K8s": []}
3289 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3292 updated_cluster_list
= []
3293 updated_v3_cluster_list
= []
3295 for vnfr_data
in db_vnfrs
.values():
3296 vca_id
= self
.get_vca_id(vnfr_data
, {})
3297 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3298 # Step 0: Prepare and set parameters
3299 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3300 vnfd_id
= vnfr_data
.get("vnfd-id")
3301 vnfd_with_id
= find_in_list(
3302 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3306 for kdud
in vnfd_with_id
["kdu"]
3307 if kdud
["name"] == kdur
["kdu-name"]
3309 namespace
= kdur
.get("k8s-namespace")
3310 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3311 if kdur
.get("helm-chart"):
3312 kdumodel
= kdur
["helm-chart"]
3313 # Default version: helm3, if helm-version is v2 assign v2
3314 k8sclustertype
= "helm-chart-v3"
3315 self
.logger
.debug("kdur: {}".format(kdur
))
3317 kdur
.get("helm-version")
3318 and kdur
.get("helm-version") == "v2"
3320 k8sclustertype
= "helm-chart"
3321 elif kdur
.get("juju-bundle"):
3322 kdumodel
= kdur
["juju-bundle"]
3323 k8sclustertype
= "juju-bundle"
3326 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3327 "juju-bundle. Maybe an old NBI version is running".format(
3328 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3331 # check if kdumodel is a file and exists
3333 vnfd_with_id
= find_in_list(
3334 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3336 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3337 if storage
: # may be not present if vnfd has not artifacts
3338 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3339 if storage
["pkg-dir"]:
3340 filename
= "{}/{}/{}s/{}".format(
3347 filename
= "{}/Scripts/{}s/{}".format(
3352 if self
.fs
.file_exists(
3353 filename
, mode
="file"
3354 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3355 kdumodel
= self
.fs
.path
+ filename
3356 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3358 except Exception: # it is not a file
3361 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3362 step
= "Synchronize repos for k8s cluster '{}'".format(
3365 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3369 k8sclustertype
== "helm-chart"
3370 and cluster_uuid
not in updated_cluster_list
3372 k8sclustertype
== "helm-chart-v3"
3373 and cluster_uuid
not in updated_v3_cluster_list
3375 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3376 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3377 cluster_uuid
=cluster_uuid
3380 if del_repo_list
or added_repo_dict
:
3381 if k8sclustertype
== "helm-chart":
3383 "_admin.helm_charts_added." + item
: None
3384 for item
in del_repo_list
3387 "_admin.helm_charts_added." + item
: name
3388 for item
, name
in added_repo_dict
.items()
3390 updated_cluster_list
.append(cluster_uuid
)
3391 elif k8sclustertype
== "helm-chart-v3":
3393 "_admin.helm_charts_v3_added." + item
: None
3394 for item
in del_repo_list
3397 "_admin.helm_charts_v3_added." + item
: name
3398 for item
, name
in added_repo_dict
.items()
3400 updated_v3_cluster_list
.append(cluster_uuid
)
3402 logging_text
+ "repos synchronized on k8s cluster "
3403 "'{}' to_delete: {}, to_add: {}".format(
3404 k8s_cluster_id
, del_repo_list
, added_repo_dict
3409 {"_id": k8s_cluster_id
},
3415 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3416 vnfr_data
["member-vnf-index-ref"],
3420 k8s_instance_info
= {
3421 "kdu-instance": None,
3422 "k8scluster-uuid": cluster_uuid
,
3423 "k8scluster-type": k8sclustertype
,
3424 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3425 "kdu-name": kdur
["kdu-name"],
3426 "kdu-model": kdumodel
,
3427 "namespace": namespace
,
3428 "kdu-deployment-name": kdu_deployment_name
,
3430 db_path
= "_admin.deployed.K8s.{}".format(index
)
3431 db_nsr_update
[db_path
] = k8s_instance_info
3432 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3433 vnfd_with_id
= find_in_list(
3434 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3436 task
= asyncio
.ensure_future(
3445 k8params
=desc_params
,
3450 self
.lcm_tasks
.register(
3454 "instantiate_KDU-{}".format(index
),
3457 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3463 except (LcmException
, asyncio
.CancelledError
):
3465 except Exception as e
:
3466 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3467 if isinstance(e
, (N2VCException
, DbException
)):
3468 self
.logger
.error(logging_text
+ msg
)
3470 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3471 raise LcmException(msg
)
3474 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3493 task_instantiation_info
,
3496 # launch instantiate_N2VC in a asyncio task and register task object
3497 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3498 # if not found, create one entry and update database
3499 # fill db_nsr._admin.deployed.VCA.<index>
3502 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3504 if "execution-environment-list" in descriptor_config
:
3505 ee_list
= descriptor_config
.get("execution-environment-list", [])
3506 elif "juju" in descriptor_config
:
3507 ee_list
= [descriptor_config
] # ns charms
3508 else: # other types as script are not supported
3511 for ee_item
in ee_list
:
3514 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3515 ee_item
.get("juju"), ee_item
.get("helm-chart")
3518 ee_descriptor_id
= ee_item
.get("id")
3519 if ee_item
.get("juju"):
3520 vca_name
= ee_item
["juju"].get("charm")
3523 if ee_item
["juju"].get("charm") is not None
3526 if ee_item
["juju"].get("cloud") == "k8s":
3527 vca_type
= "k8s_proxy_charm"
3528 elif ee_item
["juju"].get("proxy") is False:
3529 vca_type
= "native_charm"
3530 elif ee_item
.get("helm-chart"):
3531 vca_name
= ee_item
["helm-chart"]
3532 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3535 vca_type
= "helm-v3"
3538 logging_text
+ "skipping non juju neither charm configuration"
3543 for vca_index
, vca_deployed
in enumerate(
3544 db_nsr
["_admin"]["deployed"]["VCA"]
3546 if not vca_deployed
:
3549 vca_deployed
.get("member-vnf-index") == member_vnf_index
3550 and vca_deployed
.get("vdu_id") == vdu_id
3551 and vca_deployed
.get("kdu_name") == kdu_name
3552 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3553 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3557 # not found, create one.
3559 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3562 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3564 target
+= "/kdu/{}".format(kdu_name
)
3566 "target_element": target
,
3567 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3568 "member-vnf-index": member_vnf_index
,
3570 "kdu_name": kdu_name
,
3571 "vdu_count_index": vdu_index
,
3572 "operational-status": "init", # TODO revise
3573 "detailed-status": "", # TODO revise
3574 "step": "initial-deploy", # TODO revise
3576 "vdu_name": vdu_name
,
3578 "ee_descriptor_id": ee_descriptor_id
,
3582 # create VCA and configurationStatus in db
3584 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3585 "configurationStatus.{}".format(vca_index
): dict(),
3587 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3589 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3591 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3592 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3593 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3596 task_n2vc
= asyncio
.ensure_future(
3597 self
.instantiate_N2VC(
3598 logging_text
=logging_text
,
3599 vca_index
=vca_index
,
3605 vdu_index
=vdu_index
,
3606 deploy_params
=deploy_params
,
3607 config_descriptor
=descriptor_config
,
3608 base_folder
=base_folder
,
3609 nslcmop_id
=nslcmop_id
,
3613 ee_config_descriptor
=ee_item
,
3616 self
.lcm_tasks
.register(
3620 "instantiate_N2VC-{}".format(vca_index
),
3623 task_instantiation_info
[
3625 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3626 member_vnf_index
or "", vdu_id
or ""
3630 def _create_nslcmop(nsr_id
, operation
, params
):
3632 Creates a ns-lcm-opp content to be stored at database.
3633 :param nsr_id: internal id of the instance
3634 :param operation: instantiate, terminate, scale, action, ...
3635 :param params: user parameters for the operation
3636 :return: dictionary following SOL005 format
3638 # Raise exception if invalid arguments
3639 if not (nsr_id
and operation
and params
):
3641 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3648 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3649 "operationState": "PROCESSING",
3650 "statusEnteredTime": now
,
3651 "nsInstanceId": nsr_id
,
3652 "lcmOperationType": operation
,
3654 "isAutomaticInvocation": False,
3655 "operationParams": params
,
3656 "isCancelPending": False,
3658 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3659 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3664 def _format_additional_params(self
, params
):
3665 params
= params
or {}
3666 for key
, value
in params
.items():
3667 if str(value
).startswith("!!yaml "):
3668 params
[key
] = yaml
.safe_load(value
[7:])
3671 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3672 primitive
= seq
.get("name")
3673 primitive_params
= {}
3675 "member_vnf_index": vnf_index
,
3676 "primitive": primitive
,
3677 "primitive_params": primitive_params
,
3680 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3684 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3685 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3686 if op
.get("operationState") == "COMPLETED":
3687 # b. Skip sub-operation
3688 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3689 return self
.SUBOPERATION_STATUS_SKIP
3691 # c. retry executing sub-operation
3692 # The sub-operation exists, and operationState != 'COMPLETED'
3693 # Update operationState = 'PROCESSING' to indicate a retry.
3694 operationState
= "PROCESSING"
3695 detailed_status
= "In progress"
3696 self
._update
_suboperation
_status
(
3697 db_nslcmop
, op_index
, operationState
, detailed_status
3699 # Return the sub-operation index
3700 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3701 # with arguments extracted from the sub-operation
3704 # Find a sub-operation where all keys in a matching dictionary must match
3705 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3706 def _find_suboperation(self
, db_nslcmop
, match
):
3707 if db_nslcmop
and match
:
3708 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3709 for i
, op
in enumerate(op_list
):
3710 if all(op
.get(k
) == match
[k
] for k
in match
):
3712 return self
.SUBOPERATION_STATUS_NOT_FOUND
3714 # Update status for a sub-operation given its index
3715 def _update_suboperation_status(
3716 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3718 # Update DB for HA tasks
3719 q_filter
= {"_id": db_nslcmop
["_id"]}
3721 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3722 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3725 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3728 # Add sub-operation, return the index of the added sub-operation
3729 # Optionally, set operationState, detailed-status, and operationType
3730 # Status and type are currently set for 'scale' sub-operations:
3731 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3732 # 'detailed-status' : status message
3733 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3734 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3735 def _add_suboperation(
3743 mapped_primitive_params
,
3744 operationState
=None,
3745 detailed_status
=None,
3748 RO_scaling_info
=None,
3751 return self
.SUBOPERATION_STATUS_NOT_FOUND
3752 # Get the "_admin.operations" list, if it exists
3753 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3754 op_list
= db_nslcmop_admin
.get("operations")
3755 # Create or append to the "_admin.operations" list
3757 "member_vnf_index": vnf_index
,
3759 "vdu_count_index": vdu_count_index
,
3760 "primitive": primitive
,
3761 "primitive_params": mapped_primitive_params
,
3764 new_op
["operationState"] = operationState
3766 new_op
["detailed-status"] = detailed_status
3768 new_op
["lcmOperationType"] = operationType
3770 new_op
["RO_nsr_id"] = RO_nsr_id
3772 new_op
["RO_scaling_info"] = RO_scaling_info
3774 # No existing operations, create key 'operations' with current operation as first list element
3775 db_nslcmop_admin
.update({"operations": [new_op
]})
3776 op_list
= db_nslcmop_admin
.get("operations")
3778 # Existing operations, append operation to list
3779 op_list
.append(new_op
)
3781 db_nslcmop_update
= {"_admin.operations": op_list
}
3782 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3783 op_index
= len(op_list
) - 1
3786 # Helper methods for scale() sub-operations
3788 # pre-scale/post-scale:
3789 # Check for 3 different cases:
3790 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3791 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3792 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3793 def _check_or_add_scale_suboperation(
3797 vnf_config_primitive
,
3801 RO_scaling_info
=None,
3803 # Find this sub-operation
3804 if RO_nsr_id
and RO_scaling_info
:
3805 operationType
= "SCALE-RO"
3807 "member_vnf_index": vnf_index
,
3808 "RO_nsr_id": RO_nsr_id
,
3809 "RO_scaling_info": RO_scaling_info
,
3813 "member_vnf_index": vnf_index
,
3814 "primitive": vnf_config_primitive
,
3815 "primitive_params": primitive_params
,
3816 "lcmOperationType": operationType
,
3818 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3819 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3820 # a. New sub-operation
3821 # The sub-operation does not exist, add it.
3822 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3823 # The following parameters are set to None for all kind of scaling:
3825 vdu_count_index
= None
3827 if RO_nsr_id
and RO_scaling_info
:
3828 vnf_config_primitive
= None
3829 primitive_params
= None
3832 RO_scaling_info
= None
3833 # Initial status for sub-operation
3834 operationState
= "PROCESSING"
3835 detailed_status
= "In progress"
3836 # Add sub-operation for pre/post-scaling (zero or more operations)
3837 self
._add
_suboperation
(
3843 vnf_config_primitive
,
3851 return self
.SUBOPERATION_STATUS_NEW
3853 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3854 # or op_index (operationState != 'COMPLETED')
3855 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3857 # Function to return execution_environment id
3859 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3860 # TODO vdu_index_count
3861 for vca
in vca_deployed_list
:
3862 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3865 async def destroy_N2VC(
3873 exec_primitives
=True,
3878 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3879 :param logging_text:
3881 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3882 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3883 :param vca_index: index in the database _admin.deployed.VCA
3884 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3885 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3886 not executed properly
3887 :param scaling_in: True destroys the application, False destroys the model
3888 :return: None or exception
3893 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3894 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3898 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3900 # execute terminate_primitives
3902 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3903 config_descriptor
.get("terminate-config-primitive"),
3904 vca_deployed
.get("ee_descriptor_id"),
3906 vdu_id
= vca_deployed
.get("vdu_id")
3907 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3908 vdu_name
= vca_deployed
.get("vdu_name")
3909 vnf_index
= vca_deployed
.get("member-vnf-index")
3910 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3911 for seq
in terminate_primitives
:
3912 # For each sequence in list, get primitive and call _ns_execute_primitive()
3913 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3914 vnf_index
, seq
.get("name")
3916 self
.logger
.debug(logging_text
+ step
)
3917 # Create the primitive for each sequence, i.e. "primitive": "touch"
3918 primitive
= seq
.get("name")
3919 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3924 self
._add
_suboperation
(
3931 mapped_primitive_params
,
3933 # Sub-operations: Call _ns_execute_primitive() instead of action()
3935 result
, result_detail
= await self
._ns
_execute
_primitive
(
3936 vca_deployed
["ee_id"],
3938 mapped_primitive_params
,
3942 except LcmException
:
3943 # this happens when VCA is not deployed. In this case it is not needed to terminate
3945 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3946 if result
not in result_ok
:
3948 "terminate_primitive {} for vnf_member_index={} fails with "
3949 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3951 # set that this VCA do not need terminated
3952 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3956 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3959 # Delete Prometheus Jobs if any
3960 # This uses NSR_ID, so it will destroy any jobs under this index
3961 self
.db
.del_list("prometheus_jobs", {"nsr_id": db_nslcmop
["nsInstanceId"]})
3964 await self
.vca_map
[vca_type
].delete_execution_environment(
3965 vca_deployed
["ee_id"],
3966 scaling_in
=scaling_in
,
3971 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3972 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3973 namespace
= "." + db_nsr
["_id"]
3975 await self
.n2vc
.delete_namespace(
3976 namespace
=namespace
,
3977 total_timeout
=self
.timeout_charm_delete
,
3980 except N2VCNotFound
: # already deleted. Skip
3982 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3984 async def _terminate_RO(
3985 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
3988 Terminates a deployment from RO
3989 :param logging_text:
3990 :param nsr_deployed: db_nsr._admin.deployed
3993 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3994 this method will update only the index 2, but it will write on database the concatenated content of the list
3999 ro_nsr_id
= ro_delete_action
= None
4000 if nsr_deployed
and nsr_deployed
.get("RO"):
4001 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
4002 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
4005 stage
[2] = "Deleting ns from VIM."
4006 db_nsr_update
["detailed-status"] = " ".join(stage
)
4007 self
._write
_op
_status
(nslcmop_id
, stage
)
4008 self
.logger
.debug(logging_text
+ stage
[2])
4009 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4010 self
._write
_op
_status
(nslcmop_id
, stage
)
4011 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
4012 ro_delete_action
= desc
["action_id"]
4014 "_admin.deployed.RO.nsr_delete_action_id"
4015 ] = ro_delete_action
4016 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4017 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4018 if ro_delete_action
:
4019 # wait until NS is deleted from VIM
4020 stage
[2] = "Waiting ns deleted from VIM."
4021 detailed_status_old
= None
4025 + " RO_id={} ro_delete_action={}".format(
4026 ro_nsr_id
, ro_delete_action
4029 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4030 self
._write
_op
_status
(nslcmop_id
, stage
)
4032 delete_timeout
= 20 * 60 # 20 minutes
4033 while delete_timeout
> 0:
4034 desc
= await self
.RO
.show(
4036 item_id_name
=ro_nsr_id
,
4037 extra_item
="action",
4038 extra_item_id
=ro_delete_action
,
4042 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4044 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4045 if ns_status
== "ERROR":
4046 raise ROclient
.ROClientException(ns_status_info
)
4047 elif ns_status
== "BUILD":
4048 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
4049 elif ns_status
== "ACTIVE":
4050 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4051 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4056 ), "ROclient.check_action_status returns unknown {}".format(
4059 if stage
[2] != detailed_status_old
:
4060 detailed_status_old
= stage
[2]
4061 db_nsr_update
["detailed-status"] = " ".join(stage
)
4062 self
._write
_op
_status
(nslcmop_id
, stage
)
4063 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4064 await asyncio
.sleep(5, loop
=self
.loop
)
4066 else: # delete_timeout <= 0:
4067 raise ROclient
.ROClientException(
4068 "Timeout waiting ns deleted from VIM"
4071 except Exception as e
:
4072 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4074 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4076 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4077 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4078 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4080 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
4083 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4085 failed_detail
.append("delete conflict: {}".format(e
))
4088 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
4091 failed_detail
.append("delete error: {}".format(e
))
4093 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
4097 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
4098 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
4100 stage
[2] = "Deleting nsd from RO."
4101 db_nsr_update
["detailed-status"] = " ".join(stage
)
4102 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4103 self
._write
_op
_status
(nslcmop_id
, stage
)
4104 await self
.RO
.delete("nsd", ro_nsd_id
)
4106 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
4108 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4109 except Exception as e
:
4111 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4113 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4115 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
4118 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4120 failed_detail
.append(
4121 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
4123 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4125 failed_detail
.append(
4126 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
4128 self
.logger
.error(logging_text
+ failed_detail
[-1])
4130 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
4131 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
4132 if not vnf_deployed
or not vnf_deployed
["id"]:
4135 ro_vnfd_id
= vnf_deployed
["id"]
4138 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
4139 vnf_deployed
["member-vnf-index"], ro_vnfd_id
4141 db_nsr_update
["detailed-status"] = " ".join(stage
)
4142 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4143 self
._write
_op
_status
(nslcmop_id
, stage
)
4144 await self
.RO
.delete("vnfd", ro_vnfd_id
)
4146 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
4148 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
4149 except Exception as e
:
4151 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4154 "_admin.deployed.RO.vnfd.{}.id".format(index
)
4158 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4161 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4163 failed_detail
.append(
4164 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4166 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4168 failed_detail
.append(
4169 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4171 self
.logger
.error(logging_text
+ failed_detail
[-1])
4174 stage
[2] = "Error deleting from VIM"
4176 stage
[2] = "Deleted from VIM"
4177 db_nsr_update
["detailed-status"] = " ".join(stage
)
4178 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4179 self
._write
_op
_status
(nslcmop_id
, stage
)
4182 raise LcmException("; ".join(failed_detail
))
4184 async def terminate(self
, nsr_id
, nslcmop_id
):
4185 # Try to lock HA task here
4186 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4187 if not task_is_locked_by_me
:
4190 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4191 self
.logger
.debug(logging_text
+ "Enter")
4192 timeout_ns_terminate
= self
.timeout_ns_terminate
4195 operation_params
= None
4197 error_list
= [] # annotates all failed error messages
4198 db_nslcmop_update
= {}
4199 autoremove
= False # autoremove after terminated
4200 tasks_dict_info
= {}
4203 "Stage 1/3: Preparing task.",
4204 "Waiting for previous operations to terminate.",
4207 # ^ contains [stage, step, VIM-status]
4209 # wait for any previous tasks in process
4210 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4212 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4213 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4214 operation_params
= db_nslcmop
.get("operationParams") or {}
4215 if operation_params
.get("timeout_ns_terminate"):
4216 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4217 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4218 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4220 db_nsr_update
["operational-status"] = "terminating"
4221 db_nsr_update
["config-status"] = "terminating"
4222 self
._write
_ns
_status
(
4224 ns_state
="TERMINATING",
4225 current_operation
="TERMINATING",
4226 current_operation_id
=nslcmop_id
,
4227 other_update
=db_nsr_update
,
4229 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4230 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4231 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4234 stage
[1] = "Getting vnf descriptors from db."
4235 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4237 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4239 db_vnfds_from_id
= {}
4240 db_vnfds_from_member_index
= {}
4242 for vnfr
in db_vnfrs_list
:
4243 vnfd_id
= vnfr
["vnfd-id"]
4244 if vnfd_id
not in db_vnfds_from_id
:
4245 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4246 db_vnfds_from_id
[vnfd_id
] = vnfd
4247 db_vnfds_from_member_index
[
4248 vnfr
["member-vnf-index-ref"]
4249 ] = db_vnfds_from_id
[vnfd_id
]
4251 # Destroy individual execution environments when there are terminating primitives.
4252 # Rest of EE will be deleted at once
4253 # TODO - check before calling _destroy_N2VC
4254 # if not operation_params.get("skip_terminate_primitives"):#
4255 # or not vca.get("needed_terminate"):
4256 stage
[0] = "Stage 2/3 execute terminating primitives."
4257 self
.logger
.debug(logging_text
+ stage
[0])
4258 stage
[1] = "Looking execution environment that needs terminate."
4259 self
.logger
.debug(logging_text
+ stage
[1])
4261 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4262 config_descriptor
= None
4263 vca_member_vnf_index
= vca
.get("member-vnf-index")
4264 vca_id
= self
.get_vca_id(
4265 db_vnfrs_dict
.get(vca_member_vnf_index
)
4266 if vca_member_vnf_index
4270 if not vca
or not vca
.get("ee_id"):
4272 if not vca
.get("member-vnf-index"):
4274 config_descriptor
= db_nsr
.get("ns-configuration")
4275 elif vca
.get("vdu_id"):
4276 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4277 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4278 elif vca
.get("kdu_name"):
4279 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4280 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4282 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4283 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4284 vca_type
= vca
.get("type")
4285 exec_terminate_primitives
= not operation_params
.get(
4286 "skip_terminate_primitives"
4287 ) and vca
.get("needed_terminate")
4288 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4289 # pending native charms
4291 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4293 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4294 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4295 task
= asyncio
.ensure_future(
4303 exec_terminate_primitives
,
4307 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4309 # wait for pending tasks of terminate primitives
4313 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4315 error_list
= await self
._wait
_for
_tasks
(
4318 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4322 tasks_dict_info
.clear()
4324 return # raise LcmException("; ".join(error_list))
4326 # remove All execution environments at once
4327 stage
[0] = "Stage 3/3 delete all."
4329 if nsr_deployed
.get("VCA"):
4330 stage
[1] = "Deleting all execution environments."
4331 self
.logger
.debug(logging_text
+ stage
[1])
4332 vca_id
= self
.get_vca_id({}, db_nsr
)
4333 task_delete_ee
= asyncio
.ensure_future(
4335 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4336 timeout
=self
.timeout_charm_delete
,
4339 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4340 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4342 # Delete from k8scluster
4343 stage
[1] = "Deleting KDUs."
4344 self
.logger
.debug(logging_text
+ stage
[1])
4345 # print(nsr_deployed)
4346 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4347 if not kdu
or not kdu
.get("kdu-instance"):
4349 kdu_instance
= kdu
.get("kdu-instance")
4350 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4351 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4352 vca_id
= self
.get_vca_id({}, db_nsr
)
4353 task_delete_kdu_instance
= asyncio
.ensure_future(
4354 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4355 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4356 kdu_instance
=kdu_instance
,
4363 + "Unknown k8s deployment type {}".format(
4364 kdu
.get("k8scluster-type")
4369 task_delete_kdu_instance
4370 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4373 stage
[1] = "Deleting ns from VIM."
4375 task_delete_ro
= asyncio
.ensure_future(
4376 self
._terminate
_ng
_ro
(
4377 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4381 task_delete_ro
= asyncio
.ensure_future(
4383 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4386 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4388 # rest of staff will be done at finally
4391 ROclient
.ROClientException
,
4396 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4398 except asyncio
.CancelledError
:
4400 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4402 exc
= "Operation was cancelled"
4403 except Exception as e
:
4404 exc
= traceback
.format_exc()
4405 self
.logger
.critical(
4406 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4411 error_list
.append(str(exc
))
4413 # wait for pending tasks
4415 stage
[1] = "Waiting for terminate pending tasks."
4416 self
.logger
.debug(logging_text
+ stage
[1])
4417 error_list
+= await self
._wait
_for
_tasks
(
4420 timeout_ns_terminate
,
4424 stage
[1] = stage
[2] = ""
4425 except asyncio
.CancelledError
:
4426 error_list
.append("Cancelled")
4427 # TODO cancell all tasks
4428 except Exception as exc
:
4429 error_list
.append(str(exc
))
4430 # update status at database
4432 error_detail
= "; ".join(error_list
)
4433 # self.logger.error(logging_text + error_detail)
4434 error_description_nslcmop
= "{} Detail: {}".format(
4435 stage
[0], error_detail
4437 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4438 nslcmop_id
, stage
[0]
4441 db_nsr_update
["operational-status"] = "failed"
4442 db_nsr_update
["detailed-status"] = (
4443 error_description_nsr
+ " Detail: " + error_detail
4445 db_nslcmop_update
["detailed-status"] = error_detail
4446 nslcmop_operation_state
= "FAILED"
4450 error_description_nsr
= error_description_nslcmop
= None
4451 ns_state
= "NOT_INSTANTIATED"
4452 db_nsr_update
["operational-status"] = "terminated"
4453 db_nsr_update
["detailed-status"] = "Done"
4454 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4455 db_nslcmop_update
["detailed-status"] = "Done"
4456 nslcmop_operation_state
= "COMPLETED"
4459 self
._write
_ns
_status
(
4462 current_operation
="IDLE",
4463 current_operation_id
=None,
4464 error_description
=error_description_nsr
,
4465 error_detail
=error_detail
,
4466 other_update
=db_nsr_update
,
4468 self
._write
_op
_status
(
4471 error_message
=error_description_nslcmop
,
4472 operation_state
=nslcmop_operation_state
,
4473 other_update
=db_nslcmop_update
,
4475 if ns_state
== "NOT_INSTANTIATED":
4479 {"nsr-id-ref": nsr_id
},
4480 {"_admin.nsState": "NOT_INSTANTIATED"},
4482 except DbException
as e
:
4485 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4489 if operation_params
:
4490 autoremove
= operation_params
.get("autoremove", False)
4491 if nslcmop_operation_state
:
4493 await self
.msg
.aiowrite(
4498 "nslcmop_id": nslcmop_id
,
4499 "operationState": nslcmop_operation_state
,
4500 "autoremove": autoremove
,
4504 except Exception as e
:
4506 logging_text
+ "kafka_write notification Exception {}".format(e
)
4509 self
.logger
.debug(logging_text
+ "Exit")
4510 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4512 async def _wait_for_tasks(
4513 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4516 error_detail_list
= []
4518 pending_tasks
= list(created_tasks_info
.keys())
4519 num_tasks
= len(pending_tasks
)
4521 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4522 self
._write
_op
_status
(nslcmop_id
, stage
)
4523 while pending_tasks
:
4525 _timeout
= timeout
+ time_start
- time()
4526 done
, pending_tasks
= await asyncio
.wait(
4527 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4529 num_done
+= len(done
)
4530 if not done
: # Timeout
4531 for task
in pending_tasks
:
4532 new_error
= created_tasks_info
[task
] + ": Timeout"
4533 error_detail_list
.append(new_error
)
4534 error_list
.append(new_error
)
4537 if task
.cancelled():
4540 exc
= task
.exception()
4542 if isinstance(exc
, asyncio
.TimeoutError
):
4544 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4545 error_list
.append(created_tasks_info
[task
])
4546 error_detail_list
.append(new_error
)
4553 ROclient
.ROClientException
,
4559 self
.logger
.error(logging_text
+ new_error
)
4561 exc_traceback
= "".join(
4562 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4566 + created_tasks_info
[task
]
4572 logging_text
+ created_tasks_info
[task
] + ": Done"
4574 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4576 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4577 if nsr_id
: # update also nsr
4582 "errorDescription": "Error at: " + ", ".join(error_list
),
4583 "errorDetail": ". ".join(error_detail_list
),
4586 self
._write
_op
_status
(nslcmop_id
, stage
)
4587 return error_detail_list
4590 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4592 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4593 The default-value is used. If it is between < > it look for a value at instantiation_params
4594 :param primitive_desc: portion of VNFD/NSD that describes primitive
4595 :param params: Params provided by user
4596 :param instantiation_params: Instantiation params provided by user
4597 :return: a dictionary with the calculated params
4599 calculated_params
= {}
4600 for parameter
in primitive_desc
.get("parameter", ()):
4601 param_name
= parameter
["name"]
4602 if param_name
in params
:
4603 calculated_params
[param_name
] = params
[param_name
]
4604 elif "default-value" in parameter
or "value" in parameter
:
4605 if "value" in parameter
:
4606 calculated_params
[param_name
] = parameter
["value"]
4608 calculated_params
[param_name
] = parameter
["default-value"]
4610 isinstance(calculated_params
[param_name
], str)
4611 and calculated_params
[param_name
].startswith("<")
4612 and calculated_params
[param_name
].endswith(">")
4614 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4615 calculated_params
[param_name
] = instantiation_params
[
4616 calculated_params
[param_name
][1:-1]
4620 "Parameter {} needed to execute primitive {} not provided".format(
4621 calculated_params
[param_name
], primitive_desc
["name"]
4626 "Parameter {} needed to execute primitive {} not provided".format(
4627 param_name
, primitive_desc
["name"]
4631 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4632 calculated_params
[param_name
] = yaml
.safe_dump(
4633 calculated_params
[param_name
], default_flow_style
=True, width
=256
4635 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4637 ].startswith("!!yaml "):
4638 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4639 if parameter
.get("data-type") == "INTEGER":
4641 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4642 except ValueError: # error converting string to int
4644 "Parameter {} of primitive {} must be integer".format(
4645 param_name
, primitive_desc
["name"]
4648 elif parameter
.get("data-type") == "BOOLEAN":
4649 calculated_params
[param_name
] = not (
4650 (str(calculated_params
[param_name
])).lower() == "false"
4653 # add always ns_config_info if primitive name is config
4654 if primitive_desc
["name"] == "config":
4655 if "ns_config_info" in instantiation_params
:
4656 calculated_params
["ns_config_info"] = instantiation_params
[
4659 return calculated_params
4661 def _look_for_deployed_vca(
4668 ee_descriptor_id
=None,
4670 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4671 for vca
in deployed_vca
:
4674 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4677 vdu_count_index
is not None
4678 and vdu_count_index
!= vca
["vdu_count_index"]
4681 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4683 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4687 # vca_deployed not found
4689 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4690 " is not deployed".format(
4699 ee_id
= vca
.get("ee_id")
4701 "type", "lxc_proxy_charm"
4702 ) # default value for backward compatibility - proxy charm
4705 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4706 "execution environment".format(
4707 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4710 return ee_id
, vca_type
4712 async def _ns_execute_primitive(
4718 retries_interval
=30,
4725 if primitive
== "config":
4726 primitive_params
= {"params": primitive_params
}
4728 vca_type
= vca_type
or "lxc_proxy_charm"
4732 output
= await asyncio
.wait_for(
4733 self
.vca_map
[vca_type
].exec_primitive(
4735 primitive_name
=primitive
,
4736 params_dict
=primitive_params
,
4737 progress_timeout
=self
.timeout_progress_primitive
,
4738 total_timeout
=self
.timeout_primitive
,
4743 timeout
=timeout
or self
.timeout_primitive
,
4747 except asyncio
.CancelledError
:
4749 except Exception as e
: # asyncio.TimeoutError
4750 if isinstance(e
, asyncio
.TimeoutError
):
4755 "Error executing action {} on {} -> {}".format(
4760 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4762 return "FAILED", str(e
)
4764 return "COMPLETED", output
4766 except (LcmException
, asyncio
.CancelledError
):
4768 except Exception as e
:
4769 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4771 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4773 Updating the vca_status with latest juju information in nsrs record
4774 :param: nsr_id: Id of the nsr
4775 :param: nslcmop_id: Id of the nslcmop
4779 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4780 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4781 vca_id
= self
.get_vca_id({}, db_nsr
)
4782 if db_nsr
["_admin"]["deployed"]["K8s"]:
4783 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4784 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4785 await self
._on
_update
_k
8s
_db
(
4786 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4789 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4790 table
, filter = "nsrs", {"_id": nsr_id
}
4791 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4792 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4794 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4795 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4797 async def action(self
, nsr_id
, nslcmop_id
):
4798 # Try to lock HA task here
4799 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4800 if not task_is_locked_by_me
:
4803 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4804 self
.logger
.debug(logging_text
+ "Enter")
4805 # get all needed from database
4809 db_nslcmop_update
= {}
4810 nslcmop_operation_state
= None
4811 error_description_nslcmop
= None
4814 # wait for any previous tasks in process
4815 step
= "Waiting for previous operations to terminate"
4816 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4818 self
._write
_ns
_status
(
4821 current_operation
="RUNNING ACTION",
4822 current_operation_id
=nslcmop_id
,
4825 step
= "Getting information from database"
4826 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4827 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4829 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4830 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4831 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4832 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4833 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4834 primitive
= db_nslcmop
["operationParams"]["primitive"]
4835 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4836 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4837 "timeout_ns_action", self
.timeout_primitive
4841 step
= "Getting vnfr from database"
4842 db_vnfr
= self
.db
.get_one(
4843 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4845 step
= "Getting vnfd from database"
4846 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4848 # Sync filesystem before running a primitive
4849 self
.fs
.sync(db_vnfr
["vnfd-id"])
4851 step
= "Getting nsd from database"
4852 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4854 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4855 # for backward compatibility
4856 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4857 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4858 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4859 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4861 # look for primitive
4862 config_primitive_desc
= descriptor_configuration
= None
4864 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4866 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4868 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4870 descriptor_configuration
= db_nsd
.get("ns-configuration")
4872 if descriptor_configuration
and descriptor_configuration
.get(
4875 for config_primitive
in descriptor_configuration
["config-primitive"]:
4876 if config_primitive
["name"] == primitive
:
4877 config_primitive_desc
= config_primitive
4880 if not config_primitive_desc
:
4881 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4883 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4887 primitive_name
= primitive
4888 ee_descriptor_id
= None
4890 primitive_name
= config_primitive_desc
.get(
4891 "execution-environment-primitive", primitive
4893 ee_descriptor_id
= config_primitive_desc
.get(
4894 "execution-environment-ref"
4900 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4902 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4905 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4907 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4909 desc_params
= parse_yaml_strings(
4910 db_vnfr
.get("additionalParamsForVnf")
4913 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4914 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4915 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4917 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4918 actions
.add(primitive
["name"])
4919 for primitive
in kdu_configuration
.get("config-primitive", []):
4920 actions
.add(primitive
["name"])
4921 kdu_action
= True if primitive_name
in actions
else False
4923 # TODO check if ns is in a proper status
4925 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4927 # kdur and desc_params already set from before
4928 if primitive_params
:
4929 desc_params
.update(primitive_params
)
4930 # TODO Check if we will need something at vnf level
4931 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4933 kdu_name
== kdu
["kdu-name"]
4934 and kdu
["member-vnf-index"] == vnf_index
4939 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4942 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4943 msg
= "unknown k8scluster-type '{}'".format(
4944 kdu
.get("k8scluster-type")
4946 raise LcmException(msg
)
4949 "collection": "nsrs",
4950 "filter": {"_id": nsr_id
},
4951 "path": "_admin.deployed.K8s.{}".format(index
),
4955 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4957 step
= "Executing kdu {}".format(primitive_name
)
4958 if primitive_name
== "upgrade":
4959 if desc_params
.get("kdu_model"):
4960 kdu_model
= desc_params
.get("kdu_model")
4961 del desc_params
["kdu_model"]
4963 kdu_model
= kdu
.get("kdu-model")
4964 parts
= kdu_model
.split(sep
=":")
4966 kdu_model
= parts
[0]
4968 detailed_status
= await asyncio
.wait_for(
4969 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4970 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4971 kdu_instance
=kdu
.get("kdu-instance"),
4973 kdu_model
=kdu_model
,
4976 timeout
=timeout_ns_action
,
4978 timeout
=timeout_ns_action
+ 10,
4981 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4983 elif primitive_name
== "rollback":
4984 detailed_status
= await asyncio
.wait_for(
4985 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4986 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4987 kdu_instance
=kdu
.get("kdu-instance"),
4990 timeout
=timeout_ns_action
,
4992 elif primitive_name
== "status":
4993 detailed_status
= await asyncio
.wait_for(
4994 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4995 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4996 kdu_instance
=kdu
.get("kdu-instance"),
4999 timeout
=timeout_ns_action
,
5002 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
5003 kdu
["kdu-name"], nsr_id
5005 params
= self
._map
_primitive
_params
(
5006 config_primitive_desc
, primitive_params
, desc_params
5009 detailed_status
= await asyncio
.wait_for(
5010 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
5011 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5012 kdu_instance
=kdu_instance
,
5013 primitive_name
=primitive_name
,
5016 timeout
=timeout_ns_action
,
5019 timeout
=timeout_ns_action
,
5023 nslcmop_operation_state
= "COMPLETED"
5025 detailed_status
= ""
5026 nslcmop_operation_state
= "FAILED"
5028 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5029 nsr_deployed
["VCA"],
5030 member_vnf_index
=vnf_index
,
5032 vdu_count_index
=vdu_count_index
,
5033 ee_descriptor_id
=ee_descriptor_id
,
5035 for vca_index
, vca_deployed
in enumerate(
5036 db_nsr
["_admin"]["deployed"]["VCA"]
5038 if vca_deployed
.get("member-vnf-index") == vnf_index
:
5040 "collection": "nsrs",
5041 "filter": {"_id": nsr_id
},
5042 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
5046 nslcmop_operation_state
,
5048 ) = await self
._ns
_execute
_primitive
(
5050 primitive
=primitive_name
,
5051 primitive_params
=self
._map
_primitive
_params
(
5052 config_primitive_desc
, primitive_params
, desc_params
5054 timeout
=timeout_ns_action
,
5060 db_nslcmop_update
["detailed-status"] = detailed_status
5061 error_description_nslcmop
= (
5062 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
5066 + " task Done with result {} {}".format(
5067 nslcmop_operation_state
, detailed_status
5070 return # database update is called inside finally
5072 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5073 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5075 except asyncio
.CancelledError
:
5077 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5079 exc
= "Operation was cancelled"
5080 except asyncio
.TimeoutError
:
5081 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5083 except Exception as e
:
5084 exc
= traceback
.format_exc()
5085 self
.logger
.critical(
5086 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5095 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5096 nslcmop_operation_state
= "FAILED"
5098 self
._write
_ns
_status
(
5102 ], # TODO check if degraded. For the moment use previous status
5103 current_operation
="IDLE",
5104 current_operation_id
=None,
5105 # error_description=error_description_nsr,
5106 # error_detail=error_detail,
5107 other_update
=db_nsr_update
,
5110 self
._write
_op
_status
(
5113 error_message
=error_description_nslcmop
,
5114 operation_state
=nslcmop_operation_state
,
5115 other_update
=db_nslcmop_update
,
5118 if nslcmop_operation_state
:
5120 await self
.msg
.aiowrite(
5125 "nslcmop_id": nslcmop_id
,
5126 "operationState": nslcmop_operation_state
,
5130 except Exception as e
:
5132 logging_text
+ "kafka_write notification Exception {}".format(e
)
5134 self
.logger
.debug(logging_text
+ "Exit")
5135 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5136 return nslcmop_operation_state
, detailed_status
5138 async def scale(self
, nsr_id
, nslcmop_id
):
5139 # Try to lock HA task here
5140 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5141 if not task_is_locked_by_me
:
5144 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5145 stage
= ["", "", ""]
5146 tasks_dict_info
= {}
5147 # ^ stage, step, VIM progress
5148 self
.logger
.debug(logging_text
+ "Enter")
5149 # get all needed from database
5151 db_nslcmop_update
= {}
5154 # in case of error, indicates what part of scale was failed to put nsr at error status
5155 scale_process
= None
5156 old_operational_status
= ""
5157 old_config_status
= ""
5160 # wait for any previous tasks in process
5161 step
= "Waiting for previous operations to terminate"
5162 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5163 self
._write
_ns
_status
(
5166 current_operation
="SCALING",
5167 current_operation_id
=nslcmop_id
,
5170 step
= "Getting nslcmop from database"
5172 step
+ " after having waited for previous tasks to be completed"
5174 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5176 step
= "Getting nsr from database"
5177 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5178 old_operational_status
= db_nsr
["operational-status"]
5179 old_config_status
= db_nsr
["config-status"]
5181 step
= "Parsing scaling parameters"
5182 db_nsr_update
["operational-status"] = "scaling"
5183 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5184 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5186 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5188 ]["member-vnf-index"]
5189 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5191 ]["scaling-group-descriptor"]
5192 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5193 # for backward compatibility
5194 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5195 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5196 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5197 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5199 step
= "Getting vnfr from database"
5200 db_vnfr
= self
.db
.get_one(
5201 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5204 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5206 step
= "Getting vnfd from database"
5207 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5209 base_folder
= db_vnfd
["_admin"]["storage"]
5211 step
= "Getting scaling-group-descriptor"
5212 scaling_descriptor
= find_in_list(
5213 get_scaling_aspect(db_vnfd
),
5214 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5216 if not scaling_descriptor
:
5218 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5219 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5222 step
= "Sending scale order to VIM"
5223 # TODO check if ns is in a proper status
5225 if not db_nsr
["_admin"].get("scaling-group"):
5230 "_admin.scaling-group": [
5231 {"name": scaling_group
, "nb-scale-op": 0}
5235 admin_scale_index
= 0
5237 for admin_scale_index
, admin_scale_info
in enumerate(
5238 db_nsr
["_admin"]["scaling-group"]
5240 if admin_scale_info
["name"] == scaling_group
:
5241 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5243 else: # not found, set index one plus last element and add new entry with the name
5244 admin_scale_index
+= 1
5246 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5249 vca_scaling_info
= []
5250 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5251 if scaling_type
== "SCALE_OUT":
5252 if "aspect-delta-details" not in scaling_descriptor
:
5254 "Aspect delta details not fount in scaling descriptor {}".format(
5255 scaling_descriptor
["name"]
5258 # count if max-instance-count is reached
5259 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5261 scaling_info
["scaling_direction"] = "OUT"
5262 scaling_info
["vdu-create"] = {}
5263 scaling_info
["kdu-create"] = {}
5264 for delta
in deltas
:
5265 for vdu_delta
in delta
.get("vdu-delta", {}):
5266 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5267 # vdu_index also provides the number of instance of the targeted vdu
5268 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5269 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5273 additional_params
= (
5274 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5277 cloud_init_list
= []
5279 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5280 max_instance_count
= 10
5281 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5282 max_instance_count
= vdu_profile
.get(
5283 "max-number-of-instances", 10
5286 default_instance_num
= get_number_of_instances(
5289 instances_number
= vdu_delta
.get("number-of-instances", 1)
5290 nb_scale_op
+= instances_number
5292 new_instance_count
= nb_scale_op
+ default_instance_num
5293 # Control if new count is over max and vdu count is less than max.
5294 # Then assign new instance count
5295 if new_instance_count
> max_instance_count
> vdu_count
:
5296 instances_number
= new_instance_count
- max_instance_count
5298 instances_number
= instances_number
5300 if new_instance_count
> max_instance_count
:
5302 "reached the limit of {} (max-instance-count) "
5303 "scaling-out operations for the "
5304 "scaling-group-descriptor '{}'".format(
5305 nb_scale_op
, scaling_group
5308 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5310 # TODO Information of its own ip is not available because db_vnfr is not updated.
5311 additional_params
["OSM"] = get_osm_params(
5312 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5314 cloud_init_list
.append(
5315 self
._parse
_cloud
_init
(
5322 vca_scaling_info
.append(
5324 "osm_vdu_id": vdu_delta
["id"],
5325 "member-vnf-index": vnf_index
,
5327 "vdu_index": vdu_index
+ x
,
5330 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5331 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5332 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5333 kdu_name
= kdu_profile
["kdu-name"]
5334 resource_name
= kdu_profile
["resource-name"]
5336 # Might have different kdus in the same delta
5337 # Should have list for each kdu
5338 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5339 scaling_info
["kdu-create"][kdu_name
] = []
5341 kdur
= get_kdur(db_vnfr
, kdu_name
)
5342 if kdur
.get("helm-chart"):
5343 k8s_cluster_type
= "helm-chart-v3"
5344 self
.logger
.debug("kdur: {}".format(kdur
))
5346 kdur
.get("helm-version")
5347 and kdur
.get("helm-version") == "v2"
5349 k8s_cluster_type
= "helm-chart"
5350 raise NotImplementedError
5351 elif kdur
.get("juju-bundle"):
5352 k8s_cluster_type
= "juju-bundle"
5355 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5356 "juju-bundle. Maybe an old NBI version is running".format(
5357 db_vnfr
["member-vnf-index-ref"], kdu_name
5361 max_instance_count
= 10
5362 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5363 max_instance_count
= kdu_profile
.get(
5364 "max-number-of-instances", 10
5367 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5368 deployed_kdu
, _
= get_deployed_kdu(
5369 nsr_deployed
, kdu_name
, vnf_index
5371 if deployed_kdu
is None:
5373 "KDU '{}' for vnf '{}' not deployed".format(
5377 kdu_instance
= deployed_kdu
.get("kdu-instance")
5378 instance_num
= await self
.k8scluster_map
[
5380 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5381 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5382 "number-of-instances", 1
5385 # Control if new count is over max and instance_num is less than max.
5386 # Then assign max instance number to kdu replica count
5387 if kdu_replica_count
> max_instance_count
> instance_num
:
5388 kdu_replica_count
= max_instance_count
5389 if kdu_replica_count
> max_instance_count
:
5391 "reached the limit of {} (max-instance-count) "
5392 "scaling-out operations for the "
5393 "scaling-group-descriptor '{}'".format(
5394 instance_num
, scaling_group
5398 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5399 vca_scaling_info
.append(
5401 "osm_kdu_id": kdu_name
,
5402 "member-vnf-index": vnf_index
,
5404 "kdu_index": instance_num
+ x
- 1,
5407 scaling_info
["kdu-create"][kdu_name
].append(
5409 "member-vnf-index": vnf_index
,
5411 "k8s-cluster-type": k8s_cluster_type
,
5412 "resource-name": resource_name
,
5413 "scale": kdu_replica_count
,
5416 elif scaling_type
== "SCALE_IN":
5417 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5419 scaling_info
["scaling_direction"] = "IN"
5420 scaling_info
["vdu-delete"] = {}
5421 scaling_info
["kdu-delete"] = {}
5423 for delta
in deltas
:
5424 for vdu_delta
in delta
.get("vdu-delta", {}):
5425 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5426 min_instance_count
= 0
5427 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5428 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5429 min_instance_count
= vdu_profile
["min-number-of-instances"]
5431 default_instance_num
= get_number_of_instances(
5432 db_vnfd
, vdu_delta
["id"]
5434 instance_num
= vdu_delta
.get("number-of-instances", 1)
5435 nb_scale_op
-= instance_num
5437 new_instance_count
= nb_scale_op
+ default_instance_num
5439 if new_instance_count
< min_instance_count
< vdu_count
:
5440 instances_number
= min_instance_count
- new_instance_count
5442 instances_number
= instance_num
5444 if new_instance_count
< min_instance_count
:
5446 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5447 "scaling-group-descriptor '{}'".format(
5448 nb_scale_op
, scaling_group
5451 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5452 vca_scaling_info
.append(
5454 "osm_vdu_id": vdu_delta
["id"],
5455 "member-vnf-index": vnf_index
,
5457 "vdu_index": vdu_index
- 1 - x
,
5460 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5461 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5462 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5463 kdu_name
= kdu_profile
["kdu-name"]
5464 resource_name
= kdu_profile
["resource-name"]
5466 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5467 scaling_info
["kdu-delete"][kdu_name
] = []
5469 kdur
= get_kdur(db_vnfr
, kdu_name
)
5470 if kdur
.get("helm-chart"):
5471 k8s_cluster_type
= "helm-chart-v3"
5472 self
.logger
.debug("kdur: {}".format(kdur
))
5474 kdur
.get("helm-version")
5475 and kdur
.get("helm-version") == "v2"
5477 k8s_cluster_type
= "helm-chart"
5478 raise NotImplementedError
5479 elif kdur
.get("juju-bundle"):
5480 k8s_cluster_type
= "juju-bundle"
5483 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5484 "juju-bundle. Maybe an old NBI version is running".format(
5485 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5489 min_instance_count
= 0
5490 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5491 min_instance_count
= kdu_profile
["min-number-of-instances"]
5493 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5494 deployed_kdu
, _
= get_deployed_kdu(
5495 nsr_deployed
, kdu_name
, vnf_index
5497 if deployed_kdu
is None:
5499 "KDU '{}' for vnf '{}' not deployed".format(
5503 kdu_instance
= deployed_kdu
.get("kdu-instance")
5504 instance_num
= await self
.k8scluster_map
[
5506 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5507 kdu_replica_count
= instance_num
- kdu_delta
.get(
5508 "number-of-instances", 1
5511 if kdu_replica_count
< min_instance_count
< instance_num
:
5512 kdu_replica_count
= min_instance_count
5513 if kdu_replica_count
< min_instance_count
:
5515 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5516 "scaling-group-descriptor '{}'".format(
5517 instance_num
, scaling_group
5521 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5522 vca_scaling_info
.append(
5524 "osm_kdu_id": kdu_name
,
5525 "member-vnf-index": vnf_index
,
5527 "kdu_index": instance_num
- x
- 1,
5530 scaling_info
["kdu-delete"][kdu_name
].append(
5532 "member-vnf-index": vnf_index
,
5534 "k8s-cluster-type": k8s_cluster_type
,
5535 "resource-name": resource_name
,
5536 "scale": kdu_replica_count
,
5540 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5541 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5542 if scaling_info
["scaling_direction"] == "IN":
5543 for vdur
in reversed(db_vnfr
["vdur"]):
5544 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5545 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5546 scaling_info
["vdu"].append(
5548 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5549 "vdu_id": vdur
["vdu-id-ref"],
5553 for interface
in vdur
["interfaces"]:
5554 scaling_info
["vdu"][-1]["interface"].append(
5556 "name": interface
["name"],
5557 "ip_address": interface
["ip-address"],
5558 "mac_address": interface
.get("mac-address"),
5561 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5564 step
= "Executing pre-scale vnf-config-primitive"
5565 if scaling_descriptor
.get("scaling-config-action"):
5566 for scaling_config_action
in scaling_descriptor
[
5567 "scaling-config-action"
5570 scaling_config_action
.get("trigger") == "pre-scale-in"
5571 and scaling_type
== "SCALE_IN"
5573 scaling_config_action
.get("trigger") == "pre-scale-out"
5574 and scaling_type
== "SCALE_OUT"
5576 vnf_config_primitive
= scaling_config_action
[
5577 "vnf-config-primitive-name-ref"
5579 step
= db_nslcmop_update
[
5581 ] = "executing pre-scale scaling-config-action '{}'".format(
5582 vnf_config_primitive
5585 # look for primitive
5586 for config_primitive
in (
5587 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5588 ).get("config-primitive", ()):
5589 if config_primitive
["name"] == vnf_config_primitive
:
5593 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5594 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5595 "primitive".format(scaling_group
, vnf_config_primitive
)
5598 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5599 if db_vnfr
.get("additionalParamsForVnf"):
5600 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5602 scale_process
= "VCA"
5603 db_nsr_update
["config-status"] = "configuring pre-scaling"
5604 primitive_params
= self
._map
_primitive
_params
(
5605 config_primitive
, {}, vnfr_params
5608 # Pre-scale retry check: Check if this sub-operation has been executed before
5609 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5612 vnf_config_primitive
,
5616 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5617 # Skip sub-operation
5618 result
= "COMPLETED"
5619 result_detail
= "Done"
5622 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5623 vnf_config_primitive
, result
, result_detail
5627 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5628 # New sub-operation: Get index of this sub-operation
5630 len(db_nslcmop
.get("_admin", {}).get("operations"))
5635 + "vnf_config_primitive={} New sub-operation".format(
5636 vnf_config_primitive
5640 # retry: Get registered params for this existing sub-operation
5641 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5644 vnf_index
= op
.get("member_vnf_index")
5645 vnf_config_primitive
= op
.get("primitive")
5646 primitive_params
= op
.get("primitive_params")
5649 + "vnf_config_primitive={} Sub-operation retry".format(
5650 vnf_config_primitive
5653 # Execute the primitive, either with new (first-time) or registered (reintent) args
5654 ee_descriptor_id
= config_primitive
.get(
5655 "execution-environment-ref"
5657 primitive_name
= config_primitive
.get(
5658 "execution-environment-primitive", vnf_config_primitive
5660 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5661 nsr_deployed
["VCA"],
5662 member_vnf_index
=vnf_index
,
5664 vdu_count_index
=None,
5665 ee_descriptor_id
=ee_descriptor_id
,
5667 result
, result_detail
= await self
._ns
_execute
_primitive
(
5676 + "vnf_config_primitive={} Done with result {} {}".format(
5677 vnf_config_primitive
, result
, result_detail
5680 # Update operationState = COMPLETED | FAILED
5681 self
._update
_suboperation
_status
(
5682 db_nslcmop
, op_index
, result
, result_detail
5685 if result
== "FAILED":
5686 raise LcmException(result_detail
)
5687 db_nsr_update
["config-status"] = old_config_status
5688 scale_process
= None
5692 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5695 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5698 # SCALE-IN VCA - BEGIN
5699 if vca_scaling_info
:
5700 step
= db_nslcmop_update
[
5702 ] = "Deleting the execution environments"
5703 scale_process
= "VCA"
5704 for vca_info
in vca_scaling_info
:
5705 if vca_info
["type"] == "delete":
5706 member_vnf_index
= str(vca_info
["member-vnf-index"])
5708 logging_text
+ "vdu info: {}".format(vca_info
)
5710 if vca_info
.get("osm_vdu_id"):
5711 vdu_id
= vca_info
["osm_vdu_id"]
5712 vdu_index
= int(vca_info
["vdu_index"])
5715 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5716 member_vnf_index
, vdu_id
, vdu_index
5720 kdu_id
= vca_info
["osm_kdu_id"]
5723 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5724 member_vnf_index
, kdu_id
, vdu_index
5726 stage
[2] = step
= "Scaling in VCA"
5727 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5728 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5729 config_update
= db_nsr
["configurationStatus"]
5730 for vca_index
, vca
in enumerate(vca_update
):
5732 (vca
or vca
.get("ee_id"))
5733 and vca
["member-vnf-index"] == member_vnf_index
5734 and vca
["vdu_count_index"] == vdu_index
5736 if vca
.get("vdu_id"):
5737 config_descriptor
= get_configuration(
5738 db_vnfd
, vca
.get("vdu_id")
5740 elif vca
.get("kdu_name"):
5741 config_descriptor
= get_configuration(
5742 db_vnfd
, vca
.get("kdu_name")
5745 config_descriptor
= get_configuration(
5746 db_vnfd
, db_vnfd
["id"]
5748 operation_params
= (
5749 db_nslcmop
.get("operationParams") or {}
5751 exec_terminate_primitives
= not operation_params
.get(
5752 "skip_terminate_primitives"
5753 ) and vca
.get("needed_terminate")
5754 task
= asyncio
.ensure_future(
5763 exec_primitives
=exec_terminate_primitives
,
5767 timeout
=self
.timeout_charm_delete
,
5770 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5773 del vca_update
[vca_index
]
5774 del config_update
[vca_index
]
5775 # wait for pending tasks of terminate primitives
5779 + "Waiting for tasks {}".format(
5780 list(tasks_dict_info
.keys())
5783 error_list
= await self
._wait
_for
_tasks
(
5787 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5792 tasks_dict_info
.clear()
5794 raise LcmException("; ".join(error_list
))
5796 db_vca_and_config_update
= {
5797 "_admin.deployed.VCA": vca_update
,
5798 "configurationStatus": config_update
,
5801 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5803 scale_process
= None
5804 # SCALE-IN VCA - END
5807 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5808 scale_process
= "RO"
5809 if self
.ro_config
.get("ng"):
5810 await self
._scale
_ng
_ro
(
5811 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5813 scaling_info
.pop("vdu-create", None)
5814 scaling_info
.pop("vdu-delete", None)
5816 scale_process
= None
5820 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5821 scale_process
= "KDU"
5822 await self
._scale
_kdu
(
5823 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5825 scaling_info
.pop("kdu-create", None)
5826 scaling_info
.pop("kdu-delete", None)
5828 scale_process
= None
5832 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5834 # SCALE-UP VCA - BEGIN
5835 if vca_scaling_info
:
5836 step
= db_nslcmop_update
[
5838 ] = "Creating new execution environments"
5839 scale_process
= "VCA"
5840 for vca_info
in vca_scaling_info
:
5841 if vca_info
["type"] == "create":
5842 member_vnf_index
= str(vca_info
["member-vnf-index"])
5844 logging_text
+ "vdu info: {}".format(vca_info
)
5846 vnfd_id
= db_vnfr
["vnfd-ref"]
5847 if vca_info
.get("osm_vdu_id"):
5848 vdu_index
= int(vca_info
["vdu_index"])
5849 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5850 if db_vnfr
.get("additionalParamsForVnf"):
5851 deploy_params
.update(
5853 db_vnfr
["additionalParamsForVnf"].copy()
5856 descriptor_config
= get_configuration(
5857 db_vnfd
, db_vnfd
["id"]
5859 if descriptor_config
:
5864 logging_text
=logging_text
5865 + "member_vnf_index={} ".format(member_vnf_index
),
5868 nslcmop_id
=nslcmop_id
,
5874 member_vnf_index
=member_vnf_index
,
5875 vdu_index
=vdu_index
,
5877 deploy_params
=deploy_params
,
5878 descriptor_config
=descriptor_config
,
5879 base_folder
=base_folder
,
5880 task_instantiation_info
=tasks_dict_info
,
5883 vdu_id
= vca_info
["osm_vdu_id"]
5884 vdur
= find_in_list(
5885 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5887 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5888 if vdur
.get("additionalParams"):
5889 deploy_params_vdu
= parse_yaml_strings(
5890 vdur
["additionalParams"]
5893 deploy_params_vdu
= deploy_params
5894 deploy_params_vdu
["OSM"] = get_osm_params(
5895 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5897 if descriptor_config
:
5902 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5903 member_vnf_index
, vdu_id
, vdu_index
5905 stage
[2] = step
= "Scaling out VCA"
5906 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5908 logging_text
=logging_text
5909 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5910 member_vnf_index
, vdu_id
, vdu_index
5914 nslcmop_id
=nslcmop_id
,
5920 member_vnf_index
=member_vnf_index
,
5921 vdu_index
=vdu_index
,
5923 deploy_params
=deploy_params_vdu
,
5924 descriptor_config
=descriptor_config
,
5925 base_folder
=base_folder
,
5926 task_instantiation_info
=tasks_dict_info
,
5930 kdu_name
= vca_info
["osm_kdu_id"]
5931 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5932 if descriptor_config
:
5934 kdu_index
= int(vca_info
["kdu_index"])
5938 for x
in db_vnfr
["kdur"]
5939 if x
["kdu-name"] == kdu_name
5941 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5942 if kdur
.get("additionalParams"):
5943 deploy_params_kdu
= parse_yaml_strings(
5944 kdur
["additionalParams"]
5948 logging_text
=logging_text
,
5951 nslcmop_id
=nslcmop_id
,
5957 member_vnf_index
=member_vnf_index
,
5958 vdu_index
=kdu_index
,
5960 deploy_params
=deploy_params_kdu
,
5961 descriptor_config
=descriptor_config
,
5962 base_folder
=base_folder
,
5963 task_instantiation_info
=tasks_dict_info
,
5966 # SCALE-UP VCA - END
5967 scale_process
= None
5970 # execute primitive service POST-SCALING
5971 step
= "Executing post-scale vnf-config-primitive"
5972 if scaling_descriptor
.get("scaling-config-action"):
5973 for scaling_config_action
in scaling_descriptor
[
5974 "scaling-config-action"
5977 scaling_config_action
.get("trigger") == "post-scale-in"
5978 and scaling_type
== "SCALE_IN"
5980 scaling_config_action
.get("trigger") == "post-scale-out"
5981 and scaling_type
== "SCALE_OUT"
5983 vnf_config_primitive
= scaling_config_action
[
5984 "vnf-config-primitive-name-ref"
5986 step
= db_nslcmop_update
[
5988 ] = "executing post-scale scaling-config-action '{}'".format(
5989 vnf_config_primitive
5992 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5993 if db_vnfr
.get("additionalParamsForVnf"):
5994 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5996 # look for primitive
5997 for config_primitive
in (
5998 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5999 ).get("config-primitive", ()):
6000 if config_primitive
["name"] == vnf_config_primitive
:
6004 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
6005 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
6006 "config-primitive".format(
6007 scaling_group
, vnf_config_primitive
6010 scale_process
= "VCA"
6011 db_nsr_update
["config-status"] = "configuring post-scaling"
6012 primitive_params
= self
._map
_primitive
_params
(
6013 config_primitive
, {}, vnfr_params
6016 # Post-scale retry check: Check if this sub-operation has been executed before
6017 op_index
= self
._check
_or
_add
_scale
_suboperation
(
6020 vnf_config_primitive
,
6024 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
6025 # Skip sub-operation
6026 result
= "COMPLETED"
6027 result_detail
= "Done"
6030 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
6031 vnf_config_primitive
, result
, result_detail
6035 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
6036 # New sub-operation: Get index of this sub-operation
6038 len(db_nslcmop
.get("_admin", {}).get("operations"))
6043 + "vnf_config_primitive={} New sub-operation".format(
6044 vnf_config_primitive
6048 # retry: Get registered params for this existing sub-operation
6049 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
6052 vnf_index
= op
.get("member_vnf_index")
6053 vnf_config_primitive
= op
.get("primitive")
6054 primitive_params
= op
.get("primitive_params")
6057 + "vnf_config_primitive={} Sub-operation retry".format(
6058 vnf_config_primitive
6061 # Execute the primitive, either with new (first-time) or registered (reintent) args
6062 ee_descriptor_id
= config_primitive
.get(
6063 "execution-environment-ref"
6065 primitive_name
= config_primitive
.get(
6066 "execution-environment-primitive", vnf_config_primitive
6068 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6069 nsr_deployed
["VCA"],
6070 member_vnf_index
=vnf_index
,
6072 vdu_count_index
=None,
6073 ee_descriptor_id
=ee_descriptor_id
,
6075 result
, result_detail
= await self
._ns
_execute
_primitive
(
6084 + "vnf_config_primitive={} Done with result {} {}".format(
6085 vnf_config_primitive
, result
, result_detail
6088 # Update operationState = COMPLETED | FAILED
6089 self
._update
_suboperation
_status
(
6090 db_nslcmop
, op_index
, result
, result_detail
6093 if result
== "FAILED":
6094 raise LcmException(result_detail
)
6095 db_nsr_update
["config-status"] = old_config_status
6096 scale_process
= None
6101 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6102 db_nsr_update
["operational-status"] = (
6104 if old_operational_status
== "failed"
6105 else old_operational_status
6107 db_nsr_update
["config-status"] = old_config_status
6110 ROclient
.ROClientException
,
6115 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6117 except asyncio
.CancelledError
:
6119 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6121 exc
= "Operation was cancelled"
6122 except Exception as e
:
6123 exc
= traceback
.format_exc()
6124 self
.logger
.critical(
6125 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6129 self
._write
_ns
_status
(
6132 current_operation
="IDLE",
6133 current_operation_id
=None,
6136 stage
[1] = "Waiting for instantiate pending tasks."
6137 self
.logger
.debug(logging_text
+ stage
[1])
6138 exc
= await self
._wait
_for
_tasks
(
6141 self
.timeout_ns_deploy
,
6149 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6150 nslcmop_operation_state
= "FAILED"
6152 db_nsr_update
["operational-status"] = old_operational_status
6153 db_nsr_update
["config-status"] = old_config_status
6154 db_nsr_update
["detailed-status"] = ""
6156 if "VCA" in scale_process
:
6157 db_nsr_update
["config-status"] = "failed"
6158 if "RO" in scale_process
:
6159 db_nsr_update
["operational-status"] = "failed"
6162 ] = "FAILED scaling nslcmop={} {}: {}".format(
6163 nslcmop_id
, step
, exc
6166 error_description_nslcmop
= None
6167 nslcmop_operation_state
= "COMPLETED"
6168 db_nslcmop_update
["detailed-status"] = "Done"
6170 self
._write
_op
_status
(
6173 error_message
=error_description_nslcmop
,
6174 operation_state
=nslcmop_operation_state
,
6175 other_update
=db_nslcmop_update
,
6178 self
._write
_ns
_status
(
6181 current_operation
="IDLE",
6182 current_operation_id
=None,
6183 other_update
=db_nsr_update
,
6186 if nslcmop_operation_state
:
6190 "nslcmop_id": nslcmop_id
,
6191 "operationState": nslcmop_operation_state
,
6193 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6194 except Exception as e
:
6196 logging_text
+ "kafka_write notification Exception {}".format(e
)
6198 self
.logger
.debug(logging_text
+ "Exit")
6199 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6201 async def _scale_kdu(
6202 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6204 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6205 for kdu_name
in _scaling_info
:
6206 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6207 deployed_kdu
, index
= get_deployed_kdu(
6208 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6210 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6211 kdu_instance
= deployed_kdu
["kdu-instance"]
6212 scale
= int(kdu_scaling_info
["scale"])
6213 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6216 "collection": "nsrs",
6217 "filter": {"_id": nsr_id
},
6218 "path": "_admin.deployed.K8s.{}".format(index
),
6221 step
= "scaling application {}".format(
6222 kdu_scaling_info
["resource-name"]
6224 self
.logger
.debug(logging_text
+ step
)
6226 if kdu_scaling_info
["type"] == "delete":
6227 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6230 and kdu_config
.get("terminate-config-primitive")
6231 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6233 terminate_config_primitive_list
= kdu_config
.get(
6234 "terminate-config-primitive"
6236 terminate_config_primitive_list
.sort(
6237 key
=lambda val
: int(val
["seq"])
6241 terminate_config_primitive
6242 ) in terminate_config_primitive_list
:
6243 primitive_params_
= self
._map
_primitive
_params
(
6244 terminate_config_primitive
, {}, {}
6246 step
= "execute terminate config primitive"
6247 self
.logger
.debug(logging_text
+ step
)
6248 await asyncio
.wait_for(
6249 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6250 cluster_uuid
=cluster_uuid
,
6251 kdu_instance
=kdu_instance
,
6252 primitive_name
=terminate_config_primitive
["name"],
6253 params
=primitive_params_
,
6260 await asyncio
.wait_for(
6261 self
.k8scluster_map
[k8s_cluster_type
].scale(
6264 kdu_scaling_info
["resource-name"],
6267 timeout
=self
.timeout_vca_on_error
,
6270 if kdu_scaling_info
["type"] == "create":
6271 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6274 and kdu_config
.get("initial-config-primitive")
6275 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6277 initial_config_primitive_list
= kdu_config
.get(
6278 "initial-config-primitive"
6280 initial_config_primitive_list
.sort(
6281 key
=lambda val
: int(val
["seq"])
6284 for initial_config_primitive
in initial_config_primitive_list
:
6285 primitive_params_
= self
._map
_primitive
_params
(
6286 initial_config_primitive
, {}, {}
6288 step
= "execute initial config primitive"
6289 self
.logger
.debug(logging_text
+ step
)
6290 await asyncio
.wait_for(
6291 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6292 cluster_uuid
=cluster_uuid
,
6293 kdu_instance
=kdu_instance
,
6294 primitive_name
=initial_config_primitive
["name"],
6295 params
=primitive_params_
,
6302 async def _scale_ng_ro(
6303 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6305 nsr_id
= db_nslcmop
["nsInstanceId"]
6306 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6309 # read from db: vnfd's for every vnf
6312 # for each vnf in ns, read vnfd
6313 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6314 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6315 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6316 # if we haven't this vnfd, read it from db
6317 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6319 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6320 db_vnfds
.append(vnfd
)
6321 n2vc_key
= self
.n2vc
.get_public_key()
6322 n2vc_key_list
= [n2vc_key
]
6325 vdu_scaling_info
.get("vdu-create"),
6326 vdu_scaling_info
.get("vdu-delete"),
6329 # db_vnfr has been updated, update db_vnfrs to use it
6330 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6331 await self
._instantiate
_ng
_ro
(
6341 start_deploy
=time(),
6342 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6344 if vdu_scaling_info
.get("vdu-delete"):
6346 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6349 async def extract_prometheus_scrape_jobs(
6353 ee_config_descriptor
,
6358 # look if exist a file called 'prometheus*.j2' and
6359 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6363 for f
in artifact_content
6364 if f
.startswith("prometheus") and f
.endswith(".j2")
6370 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6374 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6375 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6377 vnfr_id
= vnfr_id
.replace("-", "")
6379 "JOB_NAME": vnfr_id
,
6380 "TARGET_IP": target_ip
,
6381 "EXPORTER_POD_IP": host_name
,
6382 "EXPORTER_POD_PORT": host_port
,
6384 job_list
= parse_job(job_data
, variables
)
6385 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6386 for job
in job_list
:
6388 not isinstance(job
.get("job_name"), str)
6389 or vnfr_id
not in job
["job_name"]
6391 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6392 job
["nsr_id"] = nsr_id
6393 job
["vnfr_id"] = vnfr_id
6396 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6398 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6400 :param: vim_account_id: VIM Account ID
6402 :return: (cloud_name, cloud_credential)
6404 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6405 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6407 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6409 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6411 :param: vim_account_id: VIM Account ID
6413 :return: (cloud_name, cloud_credential)
6415 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6416 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")