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
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
100 from copy
import copy
, deepcopy
101 from time
import time
102 from uuid
import uuid4
104 from random
import randint
106 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
109 class NsLcm(LcmBase
):
110 timeout_vca_on_error
= (
112 ) # Time for charm from first time at blocked,error status to mark as failed
113 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
114 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
115 timeout_charm_delete
= 10 * 60
116 timeout_primitive
= 30 * 60 # timeout for primitive execution
117 timeout_progress_primitive
= (
119 ) # timeout for some progress in a primitive execution
121 SUBOPERATION_STATUS_NOT_FOUND
= -1
122 SUBOPERATION_STATUS_NEW
= -2
123 SUBOPERATION_STATUS_SKIP
= -3
124 task_name_deploy_vca
= "Deploying VCA"
126 def __init__(self
, msg
, lcm_tasks
, config
, loop
, prometheus
=None):
128 Init, Connect to database, filesystem storage, and messaging
129 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
132 super().__init
__(msg
=msg
, logger
=logging
.getLogger("lcm.ns"))
134 self
.db
= Database().instance
.db
135 self
.fs
= Filesystem().instance
.fs
137 self
.lcm_tasks
= lcm_tasks
138 self
.timeout
= config
["timeout"]
139 self
.ro_config
= config
["ro_config"]
140 self
.ng_ro
= config
["ro_config"].get("ng")
141 self
.vca_config
= config
["VCA"].copy()
143 # create N2VC connector
144 self
.n2vc
= N2VCJujuConnector(
147 on_update_db
=self
._on
_update
_n
2vc
_db
,
152 self
.conn_helm_ee
= LCMHelmConn(
155 vca_config
=self
.vca_config
,
156 on_update_db
=self
._on
_update
_n
2vc
_db
,
159 self
.k8sclusterhelm2
= K8sHelmConnector(
160 kubectl_command
=self
.vca_config
.get("kubectlpath"),
161 helm_command
=self
.vca_config
.get("helmpath"),
168 self
.k8sclusterhelm3
= K8sHelm3Connector(
169 kubectl_command
=self
.vca_config
.get("kubectlpath"),
170 helm_command
=self
.vca_config
.get("helm3path"),
177 self
.k8sclusterjuju
= K8sJujuConnector(
178 kubectl_command
=self
.vca_config
.get("kubectlpath"),
179 juju_command
=self
.vca_config
.get("jujupath"),
182 on_update_db
=self
._on
_update
_k
8s
_db
,
187 self
.k8scluster_map
= {
188 "helm-chart": self
.k8sclusterhelm2
,
189 "helm-chart-v3": self
.k8sclusterhelm3
,
190 "chart": self
.k8sclusterhelm3
,
191 "juju-bundle": self
.k8sclusterjuju
,
192 "juju": self
.k8sclusterjuju
,
196 "lxc_proxy_charm": self
.n2vc
,
197 "native_charm": self
.n2vc
,
198 "k8s_proxy_charm": self
.n2vc
,
199 "helm": self
.conn_helm_ee
,
200 "helm-v3": self
.conn_helm_ee
,
203 self
.prometheus
= prometheus
206 self
.RO
= NgRoClient(self
.loop
, **self
.ro_config
)
209 def increment_ip_mac(ip_mac
, vm_index
=1):
210 if not isinstance(ip_mac
, str):
213 # try with ipv4 look for last dot
214 i
= ip_mac
.rfind(".")
217 return "{}{}".format(ip_mac
[:i
], int(ip_mac
[i
:]) + vm_index
)
218 # try with ipv6 or mac look for last colon. Operate in hex
219 i
= ip_mac
.rfind(":")
222 # format in hex, len can be 2 for mac or 4 for ipv6
223 return ("{}{:0" + str(len(ip_mac
) - i
) + "x}").format(
224 ip_mac
[:i
], int(ip_mac
[i
:], 16) + vm_index
230 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
232 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
235 # TODO filter RO descriptor fields...
239 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
240 db_dict
["deploymentStatus"] = ro_descriptor
241 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
243 except Exception as e
:
245 "Cannot write database RO deployment for ns={} -> {}".format(nsrs_id
, e
)
248 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
, vca_id
=None):
250 # remove last dot from path (if exists)
251 if path
.endswith("."):
254 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
255 # .format(table, filter, path, updated_data))
258 nsr_id
= filter.get("_id")
260 # read ns record from database
261 nsr
= self
.db
.get_one(table
="nsrs", q_filter
=filter)
262 current_ns_status
= nsr
.get("nsState")
264 # get vca status for NS
265 status_dict
= await self
.n2vc
.get_status(
266 namespace
="." + nsr_id
, yaml_format
=False, vca_id
=vca_id
271 db_dict
["vcaStatus"] = status_dict
272 await self
.n2vc
.update_vca_status(db_dict
["vcaStatus"], vca_id
=vca_id
)
274 # update configurationStatus for this VCA
276 vca_index
= int(path
[path
.rfind(".") + 1 :])
279 target_dict
=nsr
, key_list
=("_admin", "deployed", "VCA")
281 vca_status
= vca_list
[vca_index
].get("status")
283 configuration_status_list
= nsr
.get("configurationStatus")
284 config_status
= configuration_status_list
[vca_index
].get("status")
286 if config_status
== "BROKEN" and vca_status
!= "failed":
287 db_dict
["configurationStatus"][vca_index
] = "READY"
288 elif config_status
!= "BROKEN" and vca_status
== "failed":
289 db_dict
["configurationStatus"][vca_index
] = "BROKEN"
290 except Exception as e
:
291 # not update configurationStatus
292 self
.logger
.debug("Error updating vca_index (ignore): {}".format(e
))
294 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
295 # if nsState = 'DEGRADED' check if all is OK
297 if current_ns_status
in ("READY", "DEGRADED"):
298 error_description
= ""
300 if status_dict
.get("machines"):
301 for machine_id
in status_dict
.get("machines"):
302 machine
= status_dict
.get("machines").get(machine_id
)
303 # check machine agent-status
304 if machine
.get("agent-status"):
305 s
= machine
.get("agent-status").get("status")
308 error_description
+= (
309 "machine {} agent-status={} ; ".format(
313 # check machine instance status
314 if machine
.get("instance-status"):
315 s
= machine
.get("instance-status").get("status")
318 error_description
+= (
319 "machine {} instance-status={} ; ".format(
324 if status_dict
.get("applications"):
325 for app_id
in status_dict
.get("applications"):
326 app
= status_dict
.get("applications").get(app_id
)
327 # check application status
328 if app
.get("status"):
329 s
= app
.get("status").get("status")
332 error_description
+= (
333 "application {} status={} ; ".format(app_id
, s
)
336 if error_description
:
337 db_dict
["errorDescription"] = error_description
338 if current_ns_status
== "READY" and is_degraded
:
339 db_dict
["nsState"] = "DEGRADED"
340 if current_ns_status
== "DEGRADED" and not is_degraded
:
341 db_dict
["nsState"] = "READY"
344 self
.update_db_2("nsrs", nsr_id
, db_dict
)
346 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
348 except Exception as e
:
349 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
351 async def _on_update_k8s_db(
352 self
, cluster_uuid
, kdu_instance
, filter=None, vca_id
=None
355 Updating vca status in NSR record
356 :param cluster_uuid: UUID of a k8s cluster
357 :param kdu_instance: The unique name of the KDU instance
358 :param filter: To get nsr_id
362 # self.logger.debug("_on_update_k8s_db(cluster_uuid={}, kdu_instance={}, filter={}"
363 # .format(cluster_uuid, kdu_instance, filter))
366 nsr_id
= filter.get("_id")
368 # get vca status for NS
369 vca_status
= await self
.k8sclusterjuju
.status_kdu(
372 complete_status
=True,
378 db_dict
["vcaStatus"] = {nsr_id
: vca_status
}
380 await self
.k8sclusterjuju
.update_vca_status(
381 db_dict
["vcaStatus"],
387 self
.update_db_2("nsrs", nsr_id
, db_dict
)
389 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
391 except Exception as e
:
392 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
395 def _parse_cloud_init(cloud_init_text
, additional_params
, vnfd_id
, vdu_id
):
397 env
= Environment(undefined
=StrictUndefined
)
398 template
= env
.from_string(cloud_init_text
)
399 return template
.render(additional_params
or {})
400 except UndefinedError
as e
:
402 "Variable {} at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
403 "file, must be provided in the instantiation parameters inside the "
404 "'additionalParamsForVnf/Vdu' block".format(e
, vnfd_id
, vdu_id
)
406 except (TemplateError
, TemplateNotFound
) as e
:
408 "Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".format(
413 def _get_vdu_cloud_init_content(self
, vdu
, vnfd
):
414 cloud_init_content
= cloud_init_file
= None
416 if vdu
.get("cloud-init-file"):
417 base_folder
= vnfd
["_admin"]["storage"]
418 if base_folder
["pkg-dir"]:
419 cloud_init_file
= "{}/{}/cloud_init/{}".format(
420 base_folder
["folder"],
421 base_folder
["pkg-dir"],
422 vdu
["cloud-init-file"],
425 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
426 base_folder
["folder"],
427 vdu
["cloud-init-file"],
429 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
430 cloud_init_content
= ci_file
.read()
431 elif vdu
.get("cloud-init"):
432 cloud_init_content
= vdu
["cloud-init"]
434 return cloud_init_content
435 except FsException
as e
:
437 "Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".format(
438 vnfd
["id"], vdu
["id"], cloud_init_file
, e
442 def _get_vdu_additional_params(self
, db_vnfr
, vdu_id
):
444 vdur
for vdur
in db_vnfr
.get("vdur") if vdu_id
== vdur
["vdu-id-ref"]
446 additional_params
= vdur
.get("additionalParams")
447 return parse_yaml_strings(additional_params
)
449 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
451 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
452 :param vnfd: input vnfd
453 :param new_id: overrides vnf id if provided
454 :param additionalParams: Instantiation params for VNFs provided
455 :param nsrId: Id of the NSR
456 :return: copy of vnfd
458 vnfd_RO
= deepcopy(vnfd
)
459 # remove unused by RO configuration, monitoring, scaling and internal keys
460 vnfd_RO
.pop("_id", None)
461 vnfd_RO
.pop("_admin", None)
462 vnfd_RO
.pop("monitoring-param", None)
463 vnfd_RO
.pop("scaling-group-descriptor", None)
464 vnfd_RO
.pop("kdu", None)
465 vnfd_RO
.pop("k8s-cluster", None)
467 vnfd_RO
["id"] = new_id
469 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
470 for vdu
in get_iterable(vnfd_RO
, "vdu"):
471 vdu
.pop("cloud-init-file", None)
472 vdu
.pop("cloud-init", None)
476 def ip_profile_2_RO(ip_profile
):
477 RO_ip_profile
= deepcopy(ip_profile
)
478 if "dns-server" in RO_ip_profile
:
479 if isinstance(RO_ip_profile
["dns-server"], list):
480 RO_ip_profile
["dns-address"] = []
481 for ds
in RO_ip_profile
.pop("dns-server"):
482 RO_ip_profile
["dns-address"].append(ds
["address"])
484 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
485 if RO_ip_profile
.get("ip-version") == "ipv4":
486 RO_ip_profile
["ip-version"] = "IPv4"
487 if RO_ip_profile
.get("ip-version") == "ipv6":
488 RO_ip_profile
["ip-version"] = "IPv6"
489 if "dhcp-params" in RO_ip_profile
:
490 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
493 def _get_ro_vim_id_for_vim_account(self
, vim_account
):
494 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
495 if db_vim
["_admin"]["operationalState"] != "ENABLED":
497 "VIM={} is not available. operationalState={}".format(
498 vim_account
, db_vim
["_admin"]["operationalState"]
501 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
504 def get_ro_wim_id_for_wim_account(self
, wim_account
):
505 if isinstance(wim_account
, str):
506 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
507 if db_wim
["_admin"]["operationalState"] != "ENABLED":
509 "WIM={} is not available. operationalState={}".format(
510 wim_account
, db_wim
["_admin"]["operationalState"]
513 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
518 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None, mark_delete
=False):
520 db_vdu_push_list
= []
521 db_update
= {"_admin.modified": time()}
523 for vdu_id
, vdu_count
in vdu_create
.items():
527 for vdur
in reversed(db_vnfr
["vdur"])
528 if vdur
["vdu-id-ref"] == vdu_id
534 "Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
539 for count
in range(vdu_count
):
540 vdur_copy
= deepcopy(vdur
)
541 vdur_copy
["status"] = "BUILD"
542 vdur_copy
["status-detailed"] = None
543 vdur_copy
["ip-address"]: None
544 vdur_copy
["_id"] = str(uuid4())
545 vdur_copy
["count-index"] += count
+ 1
546 vdur_copy
["id"] = "{}-{}".format(
547 vdur_copy
["vdu-id-ref"], vdur_copy
["count-index"]
549 vdur_copy
.pop("vim_info", None)
550 for iface
in vdur_copy
["interfaces"]:
551 if iface
.get("fixed-ip"):
552 iface
["ip-address"] = self
.increment_ip_mac(
553 iface
["ip-address"], count
+ 1
556 iface
.pop("ip-address", None)
557 if iface
.get("fixed-mac"):
558 iface
["mac-address"] = self
.increment_ip_mac(
559 iface
["mac-address"], count
+ 1
562 iface
.pop("mac-address", None)
565 ) # only first vdu can be managment of vnf
566 db_vdu_push_list
.append(vdur_copy
)
567 # self.logger.debug("scale out, adding vdu={}".format(vdur_copy))
569 for vdu_id
, vdu_count
in vdu_delete
.items():
571 indexes_to_delete
= [
573 for iv
in enumerate(db_vnfr
["vdur"])
574 if iv
[1]["vdu-id-ref"] == vdu_id
578 "vdur.{}.status".format(i
): "DELETING"
579 for i
in indexes_to_delete
[-vdu_count
:]
583 # it must be deleted one by one because common.db does not allow otherwise
586 for v
in reversed(db_vnfr
["vdur"])
587 if v
["vdu-id-ref"] == vdu_id
589 for vdu
in vdus_to_delete
[:vdu_count
]:
592 {"_id": db_vnfr
["_id"]},
594 pull
={"vdur": {"_id": vdu
["_id"]}},
596 db_push
= {"vdur": db_vdu_push_list
} if db_vdu_push_list
else None
597 self
.db
.set_one("vnfrs", {"_id": db_vnfr
["_id"]}, db_update
, push_list
=db_push
)
598 # modify passed dictionary db_vnfr
599 db_vnfr_
= self
.db
.get_one("vnfrs", {"_id": db_vnfr
["_id"]})
600 db_vnfr
["vdur"] = db_vnfr_
["vdur"]
602 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
604 Updates database nsr with the RO info for the created vld
605 :param ns_update_nsr: dictionary to be filled with the updated info
606 :param db_nsr: content of db_nsr. This is also modified
607 :param nsr_desc_RO: nsr descriptor from RO
608 :return: Nothing, LcmException is raised on errors
611 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
612 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
613 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
615 vld
["vim-id"] = net_RO
.get("vim_net_id")
616 vld
["name"] = net_RO
.get("vim_name")
617 vld
["status"] = net_RO
.get("status")
618 vld
["status-detailed"] = net_RO
.get("error_msg")
619 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
623 "ns_update_nsr: Not found vld={} at RO info".format(vld
["id"])
626 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
628 for db_vnfr
in db_vnfrs
.values():
629 vnfr_update
= {"status": "ERROR"}
630 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
631 if "status" not in vdur
:
632 vdur
["status"] = "ERROR"
633 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
635 vdur
["status-detailed"] = str(error_text
)
637 "vdur.{}.status-detailed".format(vdu_index
)
639 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
640 except DbException
as e
:
641 self
.logger
.error("Cannot update vnf. {}".format(e
))
643 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
645 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
646 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
647 :param nsr_desc_RO: nsr descriptor from RO
648 :return: Nothing, LcmException is raised on errors
650 for vnf_index
, db_vnfr
in db_vnfrs
.items():
651 for vnf_RO
in nsr_desc_RO
["vnfs"]:
652 if vnf_RO
["member_vnf_index"] != vnf_index
:
655 if vnf_RO
.get("ip_address"):
656 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
[
659 elif not db_vnfr
.get("ip-address"):
660 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
661 raise LcmExceptionNoMgmtIP(
662 "ns member_vnf_index '{}' has no IP address".format(
667 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
668 vdur_RO_count_index
= 0
669 if vdur
.get("pdu-type"):
671 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
672 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
674 if vdur
["count-index"] != vdur_RO_count_index
:
675 vdur_RO_count_index
+= 1
677 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
678 if vdur_RO
.get("ip_address"):
679 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
681 vdur
["ip-address"] = None
682 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
683 vdur
["name"] = vdur_RO
.get("vim_name")
684 vdur
["status"] = vdur_RO
.get("status")
685 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
686 for ifacer
in get_iterable(vdur
, "interfaces"):
687 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
688 if ifacer
["name"] == interface_RO
.get("internal_name"):
689 ifacer
["ip-address"] = interface_RO
.get(
692 ifacer
["mac-address"] = interface_RO
.get(
698 "ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
699 "from VIM info".format(
700 vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]
703 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
707 "ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
709 vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]
713 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
714 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
715 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
717 vld
["vim-id"] = net_RO
.get("vim_net_id")
718 vld
["name"] = net_RO
.get("vim_name")
719 vld
["status"] = net_RO
.get("status")
720 vld
["status-detailed"] = net_RO
.get("error_msg")
721 vnfr_update
["vld.{}".format(vld_index
)] = vld
725 "ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
730 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
735 "ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(
740 def _get_ns_config_info(self
, nsr_id
):
742 Generates a mapping between vnf,vdu elements and the N2VC id
743 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
744 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
745 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
746 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
748 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
749 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
751 ns_config_info
= {"osm-config-mapping": mapping
}
752 for vca
in vca_deployed_list
:
753 if not vca
["member-vnf-index"]:
755 if not vca
["vdu_id"]:
756 mapping
[vca
["member-vnf-index"]] = vca
["application"]
760 vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"]
762 ] = vca
["application"]
763 return ns_config_info
765 async def _instantiate_ng_ro(
782 def get_vim_account(vim_account_id
):
784 if vim_account_id
in db_vims
:
785 return db_vims
[vim_account_id
]
786 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account_id
})
787 db_vims
[vim_account_id
] = db_vim
790 # modify target_vld info with instantiation parameters
791 def parse_vld_instantiation_params(
792 target_vim
, target_vld
, vld_params
, target_sdn
794 if vld_params
.get("ip-profile"):
795 target_vld
["vim_info"][target_vim
]["ip_profile"] = vld_params
[
798 if vld_params
.get("provider-network"):
799 target_vld
["vim_info"][target_vim
]["provider_network"] = vld_params
[
802 if "sdn-ports" in vld_params
["provider-network"] and target_sdn
:
803 target_vld
["vim_info"][target_sdn
]["sdn-ports"] = vld_params
[
806 if vld_params
.get("wimAccountId"):
807 target_wim
= "wim:{}".format(vld_params
["wimAccountId"])
808 target_vld
["vim_info"][target_wim
] = {}
809 for param
in ("vim-network-name", "vim-network-id"):
810 if vld_params
.get(param
):
811 if isinstance(vld_params
[param
], dict):
812 for vim
, vim_net
in vld_params
[param
].items():
813 other_target_vim
= "vim:" + vim
815 target_vld
["vim_info"],
816 (other_target_vim
, param
.replace("-", "_")),
819 else: # isinstance str
820 target_vld
["vim_info"][target_vim
][
821 param
.replace("-", "_")
822 ] = vld_params
[param
]
823 if vld_params
.get("common_id"):
824 target_vld
["common_id"] = vld_params
.get("common_id")
826 nslcmop_id
= db_nslcmop
["_id"]
828 "name": db_nsr
["name"],
831 "image": deepcopy(db_nsr
["image"]),
832 "flavor": deepcopy(db_nsr
["flavor"]),
833 "action_id": nslcmop_id
,
834 "cloud_init_content": {},
836 for image
in target
["image"]:
837 image
["vim_info"] = {}
838 for flavor
in target
["flavor"]:
839 flavor
["vim_info"] = {}
841 if db_nslcmop
.get("lcmOperationType") != "instantiate":
842 # get parameters of instantiation:
843 db_nslcmop_instantiate
= self
.db
.get_list(
846 "nsInstanceId": db_nslcmop
["nsInstanceId"],
847 "lcmOperationType": "instantiate",
850 ns_params
= db_nslcmop_instantiate
.get("operationParams")
852 ns_params
= db_nslcmop
.get("operationParams")
853 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
854 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
857 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
858 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
862 "mgmt-network": vld
.get("mgmt-network", False),
863 "type": vld
.get("type"),
866 "vim_network_name": vld
.get("vim-network-name"),
867 "vim_account_id": ns_params
["vimAccountId"],
871 # check if this network needs SDN assist
872 if vld
.get("pci-interfaces"):
873 db_vim
= get_vim_account(ns_params
["vimAccountId"])
874 sdnc_id
= db_vim
["config"].get("sdn-controller")
876 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
877 target_sdn
= "sdn:{}".format(sdnc_id
)
878 target_vld
["vim_info"][target_sdn
] = {
880 "target_vim": target_vim
,
882 "type": vld
.get("type"),
885 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
886 for nsd_vnf_profile
in nsd_vnf_profiles
:
887 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
888 if cp
["virtual-link-profile-id"] == vld
["id"]:
890 "member_vnf:{}.{}".format(
891 cp
["constituent-cpd-id"][0][
892 "constituent-base-element-id"
894 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
896 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
898 # check at nsd descriptor, if there is an ip-profile
900 nsd_vlp
= find_in_list(
901 get_virtual_link_profiles(nsd
),
902 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
907 and nsd_vlp
.get("virtual-link-protocol-data")
908 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
910 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
913 ip_profile_dest_data
= {}
914 if "ip-version" in ip_profile_source_data
:
915 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
918 if "cidr" in ip_profile_source_data
:
919 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
922 if "gateway-ip" in ip_profile_source_data
:
923 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
926 if "dhcp-enabled" in ip_profile_source_data
:
927 ip_profile_dest_data
["dhcp-params"] = {
928 "enabled": ip_profile_source_data
["dhcp-enabled"]
930 vld_params
["ip-profile"] = ip_profile_dest_data
932 # update vld_params with instantiation params
933 vld_instantiation_params
= find_in_list(
934 get_iterable(ns_params
, "vld"),
935 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
937 if vld_instantiation_params
:
938 vld_params
.update(vld_instantiation_params
)
939 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
940 target
["ns"]["vld"].append(target_vld
)
942 for vnfr
in db_vnfrs
.values():
944 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
946 vnf_params
= find_in_list(
947 get_iterable(ns_params
, "vnf"),
948 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
950 target_vnf
= deepcopy(vnfr
)
951 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
952 for vld
in target_vnf
.get("vld", ()):
953 # check if connected to a ns.vld, to fill target'
954 vnf_cp
= find_in_list(
955 vnfd
.get("int-virtual-link-desc", ()),
956 lambda cpd
: cpd
.get("id") == vld
["id"],
959 ns_cp
= "member_vnf:{}.{}".format(
960 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
962 if cp2target
.get(ns_cp
):
963 vld
["target"] = cp2target
[ns_cp
]
966 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
968 # check if this network needs SDN assist
970 if vld
.get("pci-interfaces"):
971 db_vim
= get_vim_account(vnfr
["vim-account-id"])
972 sdnc_id
= db_vim
["config"].get("sdn-controller")
974 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
975 target_sdn
= "sdn:{}".format(sdnc_id
)
976 vld
["vim_info"][target_sdn
] = {
978 "target_vim": target_vim
,
980 "type": vld
.get("type"),
983 # check at vnfd descriptor, if there is an ip-profile
985 vnfd_vlp
= find_in_list(
986 get_virtual_link_profiles(vnfd
),
987 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
991 and vnfd_vlp
.get("virtual-link-protocol-data")
992 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
994 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
997 ip_profile_dest_data
= {}
998 if "ip-version" in ip_profile_source_data
:
999 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
1002 if "cidr" in ip_profile_source_data
:
1003 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
1006 if "gateway-ip" in ip_profile_source_data
:
1007 ip_profile_dest_data
[
1009 ] = ip_profile_source_data
["gateway-ip"]
1010 if "dhcp-enabled" in ip_profile_source_data
:
1011 ip_profile_dest_data
["dhcp-params"] = {
1012 "enabled": ip_profile_source_data
["dhcp-enabled"]
1015 vld_params
["ip-profile"] = ip_profile_dest_data
1016 # update vld_params with instantiation params
1018 vld_instantiation_params
= find_in_list(
1019 get_iterable(vnf_params
, "internal-vld"),
1020 lambda i_vld
: i_vld
["name"] == vld
["id"],
1022 if vld_instantiation_params
:
1023 vld_params
.update(vld_instantiation_params
)
1024 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1027 for vdur
in target_vnf
.get("vdur", ()):
1028 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1029 continue # This vdu must not be created
1030 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1032 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1035 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1036 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1039 and vdu_configuration
.get("config-access")
1040 and vdu_configuration
.get("config-access").get("ssh-access")
1042 vdur
["ssh-keys"] = ssh_keys_all
1043 vdur
["ssh-access-required"] = vdu_configuration
[
1045 ]["ssh-access"]["required"]
1048 and vnf_configuration
.get("config-access")
1049 and vnf_configuration
.get("config-access").get("ssh-access")
1050 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1052 vdur
["ssh-keys"] = ssh_keys_all
1053 vdur
["ssh-access-required"] = vnf_configuration
[
1055 ]["ssh-access"]["required"]
1056 elif ssh_keys_instantiation
and find_in_list(
1057 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1059 vdur
["ssh-keys"] = ssh_keys_instantiation
1061 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1063 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1065 if vdud
.get("cloud-init-file"):
1066 vdur
["cloud-init"] = "{}:file:{}".format(
1067 vnfd
["_id"], vdud
.get("cloud-init-file")
1069 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1070 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1071 base_folder
= vnfd
["_admin"]["storage"]
1072 if base_folder
["pkg-dir"]:
1073 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1074 base_folder
["folder"],
1075 base_folder
["pkg-dir"],
1076 vdud
.get("cloud-init-file"),
1079 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
1080 base_folder
["folder"],
1081 vdud
.get("cloud-init-file"),
1083 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
1084 target
["cloud_init_content"][
1087 elif vdud
.get("cloud-init"):
1088 vdur
["cloud-init"] = "{}:vdu:{}".format(
1089 vnfd
["_id"], get_vdu_index(vnfd
, vdur
["vdu-id-ref"])
1091 # put content at target.cloul_init_content. Avoid ng_ro read vnfd descriptor
1092 target
["cloud_init_content"][vdur
["cloud-init"]] = vdud
[
1095 vdur
["additionalParams"] = vdur
.get("additionalParams") or {}
1096 deploy_params_vdu
= self
._format
_additional
_params
(
1097 vdur
.get("additionalParams") or {}
1099 deploy_params_vdu
["OSM"] = get_osm_params(
1100 vnfr
, vdur
["vdu-id-ref"], vdur
["count-index"]
1102 vdur
["additionalParams"] = deploy_params_vdu
1105 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
1106 if target_vim
not in ns_flavor
["vim_info"]:
1107 ns_flavor
["vim_info"][target_vim
] = {}
1110 # in case alternative images are provided we must check if they should be applied
1111 # for the vim_type, modify the vim_type taking into account
1112 ns_image_id
= int(vdur
["ns-image-id"])
1113 if vdur
.get("alt-image-ids"):
1114 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1115 vim_type
= db_vim
["vim_type"]
1116 for alt_image_id
in vdur
.get("alt-image-ids"):
1117 ns_alt_image
= target
["image"][int(alt_image_id
)]
1118 if vim_type
== ns_alt_image
.get("vim-type"):
1119 # must use alternative image
1121 "use alternative image id: {}".format(alt_image_id
)
1123 ns_image_id
= alt_image_id
1124 vdur
["ns-image-id"] = ns_image_id
1126 ns_image
= target
["image"][int(ns_image_id
)]
1127 if target_vim
not in ns_image
["vim_info"]:
1128 ns_image
["vim_info"][target_vim
] = {}
1130 vdur
["vim_info"] = {target_vim
: {}}
1131 # instantiation parameters
1133 # vdu_instantiation_params = next((v for v in get_iterable(vnf_params, "vdu") if v["id"] ==
1134 # vdud["id"]), None)
1135 vdur_list
.append(vdur
)
1136 target_vnf
["vdur"] = vdur_list
1137 target
["vnf"].append(target_vnf
)
1139 desc
= await self
.RO
.deploy(nsr_id
, target
)
1140 self
.logger
.debug("RO return > {}".format(desc
))
1141 action_id
= desc
["action_id"]
1142 await self
._wait
_ng
_ro
(
1143 nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
1148 "_admin.deployed.RO.operational-status": "running",
1149 "detailed-status": " ".join(stage
),
1151 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1152 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1153 self
._write
_op
_status
(nslcmop_id
, stage
)
1155 logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
)
1159 async def _wait_ng_ro(
1168 detailed_status_old
= None
1170 start_time
= start_time
or time()
1171 while time() <= start_time
+ timeout
:
1172 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
1173 self
.logger
.debug("Wait NG RO > {}".format(desc_status
))
1174 if desc_status
["status"] == "FAILED":
1175 raise NgRoException(desc_status
["details"])
1176 elif desc_status
["status"] == "BUILD":
1178 stage
[2] = "VIM: ({})".format(desc_status
["details"])
1179 elif desc_status
["status"] == "DONE":
1181 stage
[2] = "Deployed at VIM"
1184 assert False, "ROclient.check_ns_status returns unknown {}".format(
1185 desc_status
["status"]
1187 if stage
and nslcmop_id
and stage
[2] != detailed_status_old
:
1188 detailed_status_old
= stage
[2]
1189 db_nsr_update
["detailed-status"] = " ".join(stage
)
1190 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1191 self
._write
_op
_status
(nslcmop_id
, stage
)
1192 await asyncio
.sleep(15, loop
=self
.loop
)
1193 else: # timeout_ns_deploy
1194 raise NgRoException("Timeout waiting ns to deploy")
1196 async def _terminate_ng_ro(
1197 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
1202 start_deploy
= time()
1209 "action_id": nslcmop_id
,
1211 desc
= await self
.RO
.deploy(nsr_id
, target
)
1212 action_id
= desc
["action_id"]
1213 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1214 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1217 + "ns terminate action at RO. action_id={}".format(action_id
)
1221 delete_timeout
= 20 * 60 # 20 minutes
1222 await self
._wait
_ng
_ro
(
1223 nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
1226 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1227 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1229 await self
.RO
.delete(nsr_id
)
1230 except Exception as e
:
1231 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1232 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1233 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1234 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1236 logging_text
+ "RO_action_id={} already deleted".format(action_id
)
1238 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1239 failed_detail
.append("delete conflict: {}".format(e
))
1242 + "RO_action_id={} delete conflict: {}".format(action_id
, e
)
1245 failed_detail
.append("delete error: {}".format(e
))
1248 + "RO_action_id={} delete error: {}".format(action_id
, e
)
1252 stage
[2] = "Error deleting from VIM"
1254 stage
[2] = "Deleted from VIM"
1255 db_nsr_update
["detailed-status"] = " ".join(stage
)
1256 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1257 self
._write
_op
_status
(nslcmop_id
, stage
)
1260 raise LcmException("; ".join(failed_detail
))
1263 async def instantiate_RO(
1277 :param logging_text: preffix text to use at logging
1278 :param nsr_id: nsr identity
1279 :param nsd: database content of ns descriptor
1280 :param db_nsr: database content of ns record
1281 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1283 :param db_vnfds: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1284 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1285 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1286 :return: None or exception
1289 start_deploy
= time()
1290 ns_params
= db_nslcmop
.get("operationParams")
1291 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1292 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1294 timeout_ns_deploy
= self
.timeout
.get(
1295 "ns_deploy", self
.timeout_ns_deploy
1298 # Check for and optionally request placement optimization. Database will be updated if placement activated
1299 stage
[2] = "Waiting for Placement."
1300 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1301 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1302 for vnfr
in db_vnfrs
.values():
1303 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1306 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1308 return await self
._instantiate
_ng
_ro
(
1321 except Exception as e
:
1322 stage
[2] = "ERROR deploying at VIM"
1323 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1325 "Error deploying at VIM {}".format(e
),
1326 exc_info
=not isinstance(
1329 ROclient
.ROClientException
,
1338 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1340 Wait for kdu to be up, get ip address
1341 :param logging_text: prefix use for logging
1348 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1351 while nb_tries
< 360:
1352 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1356 for x
in get_iterable(db_vnfr
, "kdur")
1357 if x
.get("kdu-name") == kdu_name
1363 "Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
)
1365 if kdur
.get("status"):
1366 if kdur
["status"] in ("READY", "ENABLED"):
1367 return kdur
.get("ip-address")
1370 "target KDU={} is in error state".format(kdu_name
)
1373 await asyncio
.sleep(10, loop
=self
.loop
)
1375 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1377 async def wait_vm_up_insert_key_ro(
1378 self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None
1381 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1382 :param logging_text: prefix use for logging
1387 :param pub_key: public ssh key to inject, None to skip
1388 :param user: user to apply the public ssh key
1392 self
.logger
.debug(logging_text
+ "Starting wait_vm_up_insert_key_ro")
1396 target_vdu_id
= None
1402 if ro_retries
>= 360: # 1 hour
1404 "Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
)
1407 await asyncio
.sleep(10, loop
=self
.loop
)
1410 if not target_vdu_id
:
1411 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1413 if not vdu_id
: # for the VNF case
1414 if db_vnfr
.get("status") == "ERROR":
1416 "Cannot inject ssh-key because target VNF is in error state"
1418 ip_address
= db_vnfr
.get("ip-address")
1424 for x
in get_iterable(db_vnfr
, "vdur")
1425 if x
.get("ip-address") == ip_address
1433 for x
in get_iterable(db_vnfr
, "vdur")
1434 if x
.get("vdu-id-ref") == vdu_id
1435 and x
.get("count-index") == vdu_index
1441 not vdur
and len(db_vnfr
.get("vdur", ())) == 1
1442 ): # If only one, this should be the target vdu
1443 vdur
= db_vnfr
["vdur"][0]
1446 "Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(
1447 vnfr_id
, vdu_id
, vdu_index
1450 # New generation RO stores information at "vim_info"
1453 if vdur
.get("vim_info"):
1455 t
for t
in vdur
["vim_info"]
1456 ) # there should be only one key
1457 ng_ro_status
= vdur
["vim_info"][target_vim
].get("vim_status")
1459 vdur
.get("pdu-type")
1460 or vdur
.get("status") == "ACTIVE"
1461 or ng_ro_status
== "ACTIVE"
1463 ip_address
= vdur
.get("ip-address")
1466 target_vdu_id
= vdur
["vdu-id-ref"]
1467 elif vdur
.get("status") == "ERROR" or ng_ro_status
== "ERROR":
1469 "Cannot inject ssh-key because target VM is in error state"
1472 if not target_vdu_id
:
1475 # inject public key into machine
1476 if pub_key
and user
:
1477 self
.logger
.debug(logging_text
+ "Inserting RO key")
1478 self
.logger
.debug("SSH > PubKey > {}".format(pub_key
))
1479 if vdur
.get("pdu-type"):
1480 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1483 ro_vm_id
= "{}-{}".format(
1484 db_vnfr
["member-vnf-index-ref"], target_vdu_id
1485 ) # TODO add vdu_index
1489 "action": "inject_ssh_key",
1493 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdur
["id"]}]}],
1495 desc
= await self
.RO
.deploy(nsr_id
, target
)
1496 action_id
= desc
["action_id"]
1497 await self
._wait
_ng
_ro
(nsr_id
, action_id
, timeout
=600)
1500 # wait until NS is deployed at RO
1502 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1503 ro_nsr_id
= deep_get(
1504 db_nsrs
, ("_admin", "deployed", "RO", "nsr_id")
1508 result_dict
= await self
.RO
.create_action(
1510 item_id_name
=ro_nsr_id
,
1512 "add_public_key": pub_key
,
1517 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1518 if not result_dict
or not isinstance(result_dict
, dict):
1520 "Unknown response from RO when injecting key"
1522 for result
in result_dict
.values():
1523 if result
.get("vim_result") == 200:
1526 raise ROclient
.ROClientException(
1527 "error injecting key: {}".format(
1528 result
.get("description")
1532 except NgRoException
as e
:
1534 "Reaching max tries injecting key. Error: {}".format(e
)
1536 except ROclient
.ROClientException
as e
:
1540 + "error injecting key: {}. Retrying until {} seconds".format(
1547 "Reaching max tries injecting key. Error: {}".format(e
)
1554 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1556 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1558 my_vca
= vca_deployed_list
[vca_index
]
1559 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1560 # vdu or kdu: no dependencies
1564 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1565 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1566 configuration_status_list
= db_nsr
["configurationStatus"]
1567 for index
, vca_deployed
in enumerate(configuration_status_list
):
1568 if index
== vca_index
:
1571 if not my_vca
.get("member-vnf-index") or (
1572 vca_deployed
.get("member-vnf-index")
1573 == my_vca
.get("member-vnf-index")
1575 internal_status
= configuration_status_list
[index
].get("status")
1576 if internal_status
== "READY":
1578 elif internal_status
== "BROKEN":
1580 "Configuration aborted because dependent charm/s has failed"
1585 # no dependencies, return
1587 await asyncio
.sleep(10)
1590 raise LcmException("Configuration aborted because dependent charm/s timeout")
1592 def get_vca_id(self
, db_vnfr
: dict, db_nsr
: dict):
1595 vca_id
= deep_get(db_vnfr
, ("vca-id",))
1597 vim_account_id
= deep_get(db_nsr
, ("instantiate_params", "vimAccountId"))
1598 vca_id
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("vca")
1601 async def instantiate_N2VC(
1618 ee_config_descriptor
,
1620 nsr_id
= db_nsr
["_id"]
1621 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1622 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1623 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1624 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1626 "collection": "nsrs",
1627 "filter": {"_id": nsr_id
},
1628 "path": db_update_entry
,
1634 element_under_configuration
= nsr_id
1638 vnfr_id
= db_vnfr
["_id"]
1639 osm_config
["osm"]["vnf_id"] = vnfr_id
1641 namespace
= "{nsi}.{ns}".format(nsi
=nsi_id
if nsi_id
else "", ns
=nsr_id
)
1643 if vca_type
== "native_charm":
1646 index_number
= vdu_index
or 0
1649 element_type
= "VNF"
1650 element_under_configuration
= vnfr_id
1651 namespace
+= ".{}-{}".format(vnfr_id
, index_number
)
1653 namespace
+= ".{}-{}".format(vdu_id
, index_number
)
1654 element_type
= "VDU"
1655 element_under_configuration
= "{}-{}".format(vdu_id
, index_number
)
1656 osm_config
["osm"]["vdu_id"] = vdu_id
1658 namespace
+= ".{}".format(kdu_name
)
1659 element_type
= "KDU"
1660 element_under_configuration
= kdu_name
1661 osm_config
["osm"]["kdu_name"] = kdu_name
1664 if base_folder
["pkg-dir"]:
1665 artifact_path
= "{}/{}/{}/{}".format(
1666 base_folder
["folder"],
1667 base_folder
["pkg-dir"],
1669 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1674 artifact_path
= "{}/Scripts/{}/{}/".format(
1675 base_folder
["folder"],
1677 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1682 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1684 # get initial_config_primitive_list that applies to this element
1685 initial_config_primitive_list
= config_descriptor
.get(
1686 "initial-config-primitive"
1690 "Initial config primitive list > {}".format(
1691 initial_config_primitive_list
1695 # add config if not present for NS charm
1696 ee_descriptor_id
= ee_config_descriptor
.get("id")
1697 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1698 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1699 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1703 "Initial config primitive list #2 > {}".format(
1704 initial_config_primitive_list
1707 # n2vc_redesign STEP 3.1
1708 # find old ee_id if exists
1709 ee_id
= vca_deployed
.get("ee_id")
1711 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1712 # create or register execution environment in VCA
1713 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1715 self
._write
_configuration
_status
(
1717 vca_index
=vca_index
,
1719 element_under_configuration
=element_under_configuration
,
1720 element_type
=element_type
,
1723 step
= "create execution environment"
1724 self
.logger
.debug(logging_text
+ step
)
1728 if vca_type
== "k8s_proxy_charm":
1729 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1730 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1731 namespace
=namespace
,
1732 artifact_path
=artifact_path
,
1736 elif vca_type
== "helm" or vca_type
== "helm-v3":
1737 ee_id
, credentials
= await self
.vca_map
[
1739 ].create_execution_environment(
1740 namespace
=namespace
,
1744 artifact_path
=artifact_path
,
1748 ee_id
, credentials
= await self
.vca_map
[
1750 ].create_execution_environment(
1751 namespace
=namespace
,
1757 elif vca_type
== "native_charm":
1758 step
= "Waiting to VM being up and getting IP address"
1759 self
.logger
.debug(logging_text
+ step
)
1760 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1769 credentials
= {"hostname": rw_mgmt_ip
}
1771 username
= deep_get(
1772 config_descriptor
, ("config-access", "ssh-access", "default-user")
1774 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1775 # merged. Meanwhile let's get username from initial-config-primitive
1776 if not username
and initial_config_primitive_list
:
1777 for config_primitive
in initial_config_primitive_list
:
1778 for param
in config_primitive
.get("parameter", ()):
1779 if param
["name"] == "ssh-username":
1780 username
= param
["value"]
1784 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1785 "'config-access.ssh-access.default-user'"
1787 credentials
["username"] = username
1788 # n2vc_redesign STEP 3.2
1790 self
._write
_configuration
_status
(
1792 vca_index
=vca_index
,
1793 status
="REGISTERING",
1794 element_under_configuration
=element_under_configuration
,
1795 element_type
=element_type
,
1798 step
= "register execution environment {}".format(credentials
)
1799 self
.logger
.debug(logging_text
+ step
)
1800 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1801 credentials
=credentials
,
1802 namespace
=namespace
,
1807 # for compatibility with MON/POL modules, the need model and application name at database
1808 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1809 ee_id_parts
= ee_id
.split(".")
1810 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1811 if len(ee_id_parts
) >= 2:
1812 model_name
= ee_id_parts
[0]
1813 application_name
= ee_id_parts
[1]
1814 db_nsr_update
[db_update_entry
+ "model"] = model_name
1815 db_nsr_update
[db_update_entry
+ "application"] = application_name
1817 # n2vc_redesign STEP 3.3
1818 step
= "Install configuration Software"
1820 self
._write
_configuration
_status
(
1822 vca_index
=vca_index
,
1823 status
="INSTALLING SW",
1824 element_under_configuration
=element_under_configuration
,
1825 element_type
=element_type
,
1826 other_update
=db_nsr_update
,
1829 # TODO check if already done
1830 self
.logger
.debug(logging_text
+ step
)
1832 if vca_type
== "native_charm":
1833 config_primitive
= next(
1834 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1837 if config_primitive
:
1838 config
= self
._map
_primitive
_params
(
1839 config_primitive
, {}, deploy_params
1842 if vca_type
== "lxc_proxy_charm":
1843 if element_type
== "NS":
1844 num_units
= db_nsr
.get("config-units") or 1
1845 elif element_type
== "VNF":
1846 num_units
= db_vnfr
.get("config-units") or 1
1847 elif element_type
== "VDU":
1848 for v
in db_vnfr
["vdur"]:
1849 if vdu_id
== v
["vdu-id-ref"]:
1850 num_units
= v
.get("config-units") or 1
1852 if vca_type
!= "k8s_proxy_charm":
1853 await self
.vca_map
[vca_type
].install_configuration_sw(
1855 artifact_path
=artifact_path
,
1858 num_units
=num_units
,
1863 # write in db flag of configuration_sw already installed
1865 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1868 # add relations for this VCA (wait for other peers related with this VCA)
1869 await self
._add
_vca
_relations
(
1870 logging_text
=logging_text
,
1873 vca_index
=vca_index
,
1876 # if SSH access is required, then get execution environment SSH public
1877 # if native charm we have waited already to VM be UP
1878 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1881 # self.logger.debug("get ssh key block")
1883 config_descriptor
, ("config-access", "ssh-access", "required")
1885 # self.logger.debug("ssh key needed")
1886 # Needed to inject a ssh key
1889 ("config-access", "ssh-access", "default-user"),
1891 step
= "Install configuration Software, getting public ssh key"
1892 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1893 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1896 step
= "Insert public key into VM user={} ssh_key={}".format(
1900 # self.logger.debug("no need to get ssh key")
1901 step
= "Waiting to VM being up and getting IP address"
1902 self
.logger
.debug(logging_text
+ step
)
1904 # n2vc_redesign STEP 5.1
1905 # wait for RO (ip-address) Insert pub_key into VM
1908 rw_mgmt_ip
= await self
.wait_kdu_up(
1909 logging_text
, nsr_id
, vnfr_id
, kdu_name
1912 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1922 rw_mgmt_ip
= None # This is for a NS configuration
1924 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1926 # store rw_mgmt_ip in deploy params for later replacement
1927 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1929 # n2vc_redesign STEP 6 Execute initial config primitive
1930 step
= "execute initial config primitive"
1932 # wait for dependent primitives execution (NS -> VNF -> VDU)
1933 if initial_config_primitive_list
:
1934 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1936 # stage, in function of element type: vdu, kdu, vnf or ns
1937 my_vca
= vca_deployed_list
[vca_index
]
1938 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1940 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1941 elif my_vca
.get("member-vnf-index"):
1943 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1946 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1948 self
._write
_configuration
_status
(
1949 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1952 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1954 check_if_terminated_needed
= True
1955 for initial_config_primitive
in initial_config_primitive_list
:
1956 # adding information on the vca_deployed if it is a NS execution environment
1957 if not vca_deployed
["member-vnf-index"]:
1958 deploy_params
["ns_config_info"] = json
.dumps(
1959 self
._get
_ns
_config
_info
(nsr_id
)
1961 # TODO check if already done
1962 primitive_params_
= self
._map
_primitive
_params
(
1963 initial_config_primitive
, {}, deploy_params
1966 step
= "execute primitive '{}' params '{}'".format(
1967 initial_config_primitive
["name"], primitive_params_
1969 self
.logger
.debug(logging_text
+ step
)
1970 await self
.vca_map
[vca_type
].exec_primitive(
1972 primitive_name
=initial_config_primitive
["name"],
1973 params_dict
=primitive_params_
,
1978 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1979 if check_if_terminated_needed
:
1980 if config_descriptor
.get("terminate-config-primitive"):
1982 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
1984 check_if_terminated_needed
= False
1986 # TODO register in database that primitive is done
1988 # STEP 7 Configure metrics
1989 if vca_type
== "helm" or vca_type
== "helm-v3":
1990 prometheus_jobs
= await self
.add_prometheus_metrics(
1992 artifact_path
=artifact_path
,
1993 ee_config_descriptor
=ee_config_descriptor
,
1996 target_ip
=rw_mgmt_ip
,
2002 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
2005 step
= "instantiated at VCA"
2006 self
.logger
.debug(logging_text
+ step
)
2008 self
._write
_configuration
_status
(
2009 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
2012 except Exception as e
: # TODO not use Exception but N2VC exception
2013 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
2015 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
2018 "Exception while {} : {}".format(step
, e
), exc_info
=True
2020 self
._write
_configuration
_status
(
2021 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
2023 raise LcmException("{} {}".format(step
, e
)) from e
2025 def _write_ns_status(
2029 current_operation
: str,
2030 current_operation_id
: str,
2031 error_description
: str = None,
2032 error_detail
: str = None,
2033 other_update
: dict = None,
2036 Update db_nsr fields.
2039 :param current_operation:
2040 :param current_operation_id:
2041 :param error_description:
2042 :param error_detail:
2043 :param other_update: Other required changes at database if provided, will be cleared
2047 db_dict
= other_update
or {}
2050 ] = current_operation_id
# for backward compatibility
2051 db_dict
["_admin.current-operation"] = current_operation_id
2052 db_dict
["_admin.operation-type"] = (
2053 current_operation
if current_operation
!= "IDLE" else None
2055 db_dict
["currentOperation"] = current_operation
2056 db_dict
["currentOperationID"] = current_operation_id
2057 db_dict
["errorDescription"] = error_description
2058 db_dict
["errorDetail"] = error_detail
2061 db_dict
["nsState"] = ns_state
2062 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2063 except DbException
as e
:
2064 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2066 def _write_op_status(
2070 error_message
: str = None,
2071 queuePosition
: int = 0,
2072 operation_state
: str = None,
2073 other_update
: dict = None,
2076 db_dict
= other_update
or {}
2077 db_dict
["queuePosition"] = queuePosition
2078 if isinstance(stage
, list):
2079 db_dict
["stage"] = stage
[0]
2080 db_dict
["detailed-status"] = " ".join(stage
)
2081 elif stage
is not None:
2082 db_dict
["stage"] = str(stage
)
2084 if error_message
is not None:
2085 db_dict
["errorMessage"] = error_message
2086 if operation_state
is not None:
2087 db_dict
["operationState"] = operation_state
2088 db_dict
["statusEnteredTime"] = time()
2089 self
.update_db_2("nslcmops", op_id
, db_dict
)
2090 except DbException
as e
:
2092 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2095 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2097 nsr_id
= db_nsr
["_id"]
2098 # configurationStatus
2099 config_status
= db_nsr
.get("configurationStatus")
2102 "configurationStatus.{}.status".format(index
): status
2103 for index
, v
in enumerate(config_status
)
2107 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2109 except DbException
as e
:
2111 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2114 def _write_configuration_status(
2119 element_under_configuration
: str = None,
2120 element_type
: str = None,
2121 other_update
: dict = None,
2124 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2125 # .format(vca_index, status))
2128 db_path
= "configurationStatus.{}.".format(vca_index
)
2129 db_dict
= other_update
or {}
2131 db_dict
[db_path
+ "status"] = status
2132 if element_under_configuration
:
2134 db_path
+ "elementUnderConfiguration"
2135 ] = element_under_configuration
2137 db_dict
[db_path
+ "elementType"] = element_type
2138 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2139 except DbException
as e
:
2141 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2142 status
, nsr_id
, vca_index
, e
2146 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2148 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2149 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2150 Database is used because the result can be obtained from a different LCM worker in case of HA.
2151 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2152 :param db_nslcmop: database content of nslcmop
2153 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2154 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2155 computed 'vim-account-id'
2158 nslcmop_id
= db_nslcmop
["_id"]
2159 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2160 if placement_engine
== "PLA":
2162 logging_text
+ "Invoke and wait for placement optimization"
2164 await self
.msg
.aiowrite(
2165 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2167 db_poll_interval
= 5
2168 wait
= db_poll_interval
* 10
2170 while not pla_result
and wait
>= 0:
2171 await asyncio
.sleep(db_poll_interval
)
2172 wait
-= db_poll_interval
2173 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2174 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2178 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2181 for pla_vnf
in pla_result
["vnf"]:
2182 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2183 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2188 {"_id": vnfr
["_id"]},
2189 {"vim-account-id": pla_vnf
["vimAccountId"]},
2192 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2195 def update_nsrs_with_pla_result(self
, params
):
2197 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2199 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2201 except Exception as e
:
2202 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2204 async def instantiate(self
, nsr_id
, nslcmop_id
):
2207 :param nsr_id: ns instance to deploy
2208 :param nslcmop_id: operation to run
2212 # Try to lock HA task here
2213 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2214 if not task_is_locked_by_me
:
2216 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2220 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2221 self
.logger
.debug(logging_text
+ "Enter")
2223 # get all needed from database
2225 # database nsrs record
2228 # database nslcmops record
2231 # update operation on nsrs
2233 # update operation on nslcmops
2234 db_nslcmop_update
= {}
2236 nslcmop_operation_state
= None
2237 db_vnfrs
= {} # vnf's info indexed by member-index
2239 tasks_dict_info
= {} # from task to info text
2243 "Stage 1/5: preparation of the environment.",
2244 "Waiting for previous operations to terminate.",
2247 # ^ stage, step, VIM progress
2249 # wait for any previous tasks in process
2250 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2252 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2253 stage
[1] = "Reading from database."
2254 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2255 db_nsr_update
["detailed-status"] = "creating"
2256 db_nsr_update
["operational-status"] = "init"
2257 self
._write
_ns
_status
(
2259 ns_state
="BUILDING",
2260 current_operation
="INSTANTIATING",
2261 current_operation_id
=nslcmop_id
,
2262 other_update
=db_nsr_update
,
2264 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2266 # read from db: operation
2267 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2268 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2269 ns_params
= db_nslcmop
.get("operationParams")
2270 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2271 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2273 timeout_ns_deploy
= self
.timeout
.get(
2274 "ns_deploy", self
.timeout_ns_deploy
2278 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2279 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2280 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2281 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2282 self
.fs
.sync(db_nsr
["nsd-id"])
2284 # nsr_name = db_nsr["name"] # TODO short-name??
2286 # read from db: vnf's of this ns
2287 stage
[1] = "Getting vnfrs from db."
2288 self
.logger
.debug(logging_text
+ stage
[1])
2289 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2291 # read from db: vnfd's for every vnf
2292 db_vnfds
= [] # every vnfd data
2294 # for each vnf in ns, read vnfd
2295 for vnfr
in db_vnfrs_list
:
2296 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2297 vnfd_id
= vnfr
["vnfd-id"]
2298 vnfd_ref
= vnfr
["vnfd-ref"]
2299 self
.fs
.sync(vnfd_id
)
2301 # if we haven't this vnfd, read it from db
2302 if vnfd_id
not in db_vnfds
:
2304 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2307 self
.logger
.debug(logging_text
+ stage
[1])
2308 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2311 db_vnfds
.append(vnfd
)
2313 # Get or generates the _admin.deployed.VCA list
2314 vca_deployed_list
= None
2315 if db_nsr
["_admin"].get("deployed"):
2316 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2317 if vca_deployed_list
is None:
2318 vca_deployed_list
= []
2319 configuration_status_list
= []
2320 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2321 db_nsr_update
["configurationStatus"] = configuration_status_list
2322 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2323 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2324 elif isinstance(vca_deployed_list
, dict):
2325 # maintain backward compatibility. Change a dict to list at database
2326 vca_deployed_list
= list(vca_deployed_list
.values())
2327 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2328 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2331 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2333 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2334 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2336 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2337 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2338 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2340 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2343 # n2vc_redesign STEP 2 Deploy Network Scenario
2344 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2345 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2347 stage
[1] = "Deploying KDUs."
2348 # self.logger.debug(logging_text + "Before deploy_kdus")
2349 # Call to deploy_kdus in case exists the "vdu:kdu" param
2350 await self
.deploy_kdus(
2351 logging_text
=logging_text
,
2353 nslcmop_id
=nslcmop_id
,
2356 task_instantiation_info
=tasks_dict_info
,
2359 stage
[1] = "Getting VCA public key."
2360 # n2vc_redesign STEP 1 Get VCA public ssh-key
2361 # feature 1429. Add n2vc public key to needed VMs
2362 n2vc_key
= self
.n2vc
.get_public_key()
2363 n2vc_key_list
= [n2vc_key
]
2364 if self
.vca_config
.get("public_key"):
2365 n2vc_key_list
.append(self
.vca_config
["public_key"])
2367 stage
[1] = "Deploying NS at VIM."
2368 task_ro
= asyncio
.ensure_future(
2369 self
.instantiate_RO(
2370 logging_text
=logging_text
,
2374 db_nslcmop
=db_nslcmop
,
2377 n2vc_key_list
=n2vc_key_list
,
2381 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2382 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2384 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2385 stage
[1] = "Deploying Execution Environments."
2386 self
.logger
.debug(logging_text
+ stage
[1])
2388 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2389 for vnf_profile
in get_vnf_profiles(nsd
):
2390 vnfd_id
= vnf_profile
["vnfd-id"]
2391 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2392 member_vnf_index
= str(vnf_profile
["id"])
2393 db_vnfr
= db_vnfrs
[member_vnf_index
]
2394 base_folder
= vnfd
["_admin"]["storage"]
2400 # Get additional parameters
2401 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2402 if db_vnfr
.get("additionalParamsForVnf"):
2403 deploy_params
.update(
2404 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2407 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2408 if descriptor_config
:
2410 logging_text
=logging_text
2411 + "member_vnf_index={} ".format(member_vnf_index
),
2414 nslcmop_id
=nslcmop_id
,
2420 member_vnf_index
=member_vnf_index
,
2421 vdu_index
=vdu_index
,
2423 deploy_params
=deploy_params
,
2424 descriptor_config
=descriptor_config
,
2425 base_folder
=base_folder
,
2426 task_instantiation_info
=tasks_dict_info
,
2430 # Deploy charms for each VDU that supports one.
2431 for vdud
in get_vdu_list(vnfd
):
2433 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2434 vdur
= find_in_list(
2435 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2438 if vdur
.get("additionalParams"):
2439 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2441 deploy_params_vdu
= deploy_params
2442 deploy_params_vdu
["OSM"] = get_osm_params(
2443 db_vnfr
, vdu_id
, vdu_count_index
=0
2445 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2447 self
.logger
.debug("VDUD > {}".format(vdud
))
2449 "Descriptor config > {}".format(descriptor_config
)
2451 if descriptor_config
:
2454 for vdu_index
in range(vdud_count
):
2455 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2457 logging_text
=logging_text
2458 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2459 member_vnf_index
, vdu_id
, vdu_index
2463 nslcmop_id
=nslcmop_id
,
2469 member_vnf_index
=member_vnf_index
,
2470 vdu_index
=vdu_index
,
2472 deploy_params
=deploy_params_vdu
,
2473 descriptor_config
=descriptor_config
,
2474 base_folder
=base_folder
,
2475 task_instantiation_info
=tasks_dict_info
,
2478 for kdud
in get_kdu_list(vnfd
):
2479 kdu_name
= kdud
["name"]
2480 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2481 if descriptor_config
:
2486 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2488 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2489 if kdur
.get("additionalParams"):
2490 deploy_params_kdu
= parse_yaml_strings(
2491 kdur
["additionalParams"]
2495 logging_text
=logging_text
,
2498 nslcmop_id
=nslcmop_id
,
2504 member_vnf_index
=member_vnf_index
,
2505 vdu_index
=vdu_index
,
2507 deploy_params
=deploy_params_kdu
,
2508 descriptor_config
=descriptor_config
,
2509 base_folder
=base_folder
,
2510 task_instantiation_info
=tasks_dict_info
,
2514 # Check if this NS has a charm configuration
2515 descriptor_config
= nsd
.get("ns-configuration")
2516 if descriptor_config
and descriptor_config
.get("juju"):
2519 member_vnf_index
= None
2525 # Get additional parameters
2526 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2527 if db_nsr
.get("additionalParamsForNs"):
2528 deploy_params
.update(
2529 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2531 base_folder
= nsd
["_admin"]["storage"]
2533 logging_text
=logging_text
,
2536 nslcmop_id
=nslcmop_id
,
2542 member_vnf_index
=member_vnf_index
,
2543 vdu_index
=vdu_index
,
2545 deploy_params
=deploy_params
,
2546 descriptor_config
=descriptor_config
,
2547 base_folder
=base_folder
,
2548 task_instantiation_info
=tasks_dict_info
,
2552 # rest of staff will be done at finally
2555 ROclient
.ROClientException
,
2561 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2564 except asyncio
.CancelledError
:
2566 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2568 exc
= "Operation was cancelled"
2569 except Exception as e
:
2570 exc
= traceback
.format_exc()
2571 self
.logger
.critical(
2572 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2577 error_list
.append(str(exc
))
2579 # wait for pending tasks
2581 stage
[1] = "Waiting for instantiate pending tasks."
2582 self
.logger
.debug(logging_text
+ stage
[1])
2583 error_list
+= await self
._wait
_for
_tasks
(
2591 stage
[1] = stage
[2] = ""
2592 except asyncio
.CancelledError
:
2593 error_list
.append("Cancelled")
2594 # TODO cancel all tasks
2595 except Exception as exc
:
2596 error_list
.append(str(exc
))
2598 # update operation-status
2599 db_nsr_update
["operational-status"] = "running"
2600 # let's begin with VCA 'configured' status (later we can change it)
2601 db_nsr_update
["config-status"] = "configured"
2602 for task
, task_name
in tasks_dict_info
.items():
2603 if not task
.done() or task
.cancelled() or task
.exception():
2604 if task_name
.startswith(self
.task_name_deploy_vca
):
2605 # A N2VC task is pending
2606 db_nsr_update
["config-status"] = "failed"
2608 # RO or KDU task is pending
2609 db_nsr_update
["operational-status"] = "failed"
2611 # update status at database
2613 error_detail
= ". ".join(error_list
)
2614 self
.logger
.error(logging_text
+ error_detail
)
2615 error_description_nslcmop
= "{} Detail: {}".format(
2616 stage
[0], error_detail
2618 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2619 nslcmop_id
, stage
[0]
2622 db_nsr_update
["detailed-status"] = (
2623 error_description_nsr
+ " Detail: " + error_detail
2625 db_nslcmop_update
["detailed-status"] = error_detail
2626 nslcmop_operation_state
= "FAILED"
2630 error_description_nsr
= error_description_nslcmop
= None
2632 db_nsr_update
["detailed-status"] = "Done"
2633 db_nslcmop_update
["detailed-status"] = "Done"
2634 nslcmop_operation_state
= "COMPLETED"
2637 self
._write
_ns
_status
(
2640 current_operation
="IDLE",
2641 current_operation_id
=None,
2642 error_description
=error_description_nsr
,
2643 error_detail
=error_detail
,
2644 other_update
=db_nsr_update
,
2646 self
._write
_op
_status
(
2649 error_message
=error_description_nslcmop
,
2650 operation_state
=nslcmop_operation_state
,
2651 other_update
=db_nslcmop_update
,
2654 if nslcmop_operation_state
:
2656 await self
.msg
.aiowrite(
2661 "nslcmop_id": nslcmop_id
,
2662 "operationState": nslcmop_operation_state
,
2666 except Exception as e
:
2668 logging_text
+ "kafka_write notification Exception {}".format(e
)
2671 self
.logger
.debug(logging_text
+ "Exit")
2672 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2674 def _get_vnfd(self
, vnfd_id
: str, cached_vnfds
: Dict
[str, Any
]):
2675 if vnfd_id
not in cached_vnfds
:
2676 cached_vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
2677 return cached_vnfds
[vnfd_id
]
2679 def _get_vnfr(self
, nsr_id
: str, vnf_profile_id
: str, cached_vnfrs
: Dict
[str, Any
]):
2680 if vnf_profile_id
not in cached_vnfrs
:
2681 cached_vnfrs
[vnf_profile_id
] = self
.db
.get_one(
2684 "member-vnf-index-ref": vnf_profile_id
,
2685 "nsr-id-ref": nsr_id
,
2688 return cached_vnfrs
[vnf_profile_id
]
2690 def _is_deployed_vca_in_relation(
2691 self
, vca
: DeployedVCA
, relation
: Relation
2694 for endpoint
in (relation
.provider
, relation
.requirer
):
2695 if endpoint
["kdu-resource-profile-id"]:
2698 vca
.vnf_profile_id
== endpoint
.vnf_profile_id
2699 and vca
.vdu_profile_id
== endpoint
.vdu_profile_id
2700 and vca
.execution_environment_ref
== endpoint
.execution_environment_ref
2706 def _update_ee_relation_data_with_implicit_data(
2707 self
, nsr_id
, nsd
, ee_relation_data
, cached_vnfds
, vnf_profile_id
: str = None
2709 ee_relation_data
= safe_get_ee_relation(
2710 nsr_id
, ee_relation_data
, vnf_profile_id
=vnf_profile_id
2712 ee_relation_level
= EELevel
.get_level(ee_relation_data
)
2713 if (ee_relation_level
in (EELevel
.VNF
, EELevel
.VDU
)) and not ee_relation_data
[
2714 "execution-environment-ref"
2716 vnf_profile
= get_vnf_profile(nsd
, ee_relation_data
["vnf-profile-id"])
2717 vnfd_id
= vnf_profile
["vnfd-id"]
2718 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2721 if ee_relation_level
== EELevel
.VNF
2722 else ee_relation_data
["vdu-profile-id"]
2724 ee
= get_juju_ee_ref(db_vnfd
, entity_id
)
2727 f
"not execution environments found for ee_relation {ee_relation_data}"
2729 ee_relation_data
["execution-environment-ref"] = ee
["id"]
2730 return ee_relation_data
2732 def _get_ns_relations(
2735 nsd
: Dict
[str, Any
],
2737 cached_vnfds
: Dict
[str, Any
],
2740 db_ns_relations
= get_ns_configuration_relation_list(nsd
)
2741 for r
in db_ns_relations
:
2742 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2743 nsr_id
, nsd
, r
["provider"], cached_vnfds
2745 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2746 nsr_id
, nsd
, r
["requirer"], cached_vnfds
2748 provider
= EERelation(relation_provider
)
2749 requirer
= EERelation(relation_requirer
)
2750 relation
= Relation(r
["name"], provider
, requirer
)
2751 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2753 relations
.append(relation
)
2756 def _get_vnf_relations(
2759 nsd
: Dict
[str, Any
],
2761 cached_vnfds
: Dict
[str, Any
],
2764 vnf_profile
= get_vnf_profile(nsd
, vca
.vnf_profile_id
)
2765 vnf_profile_id
= vnf_profile
["id"]
2766 vnfd_id
= vnf_profile
["vnfd-id"]
2767 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2768 db_vnf_relations
= get_relation_list(db_vnfd
, vnfd_id
)
2769 for r
in db_vnf_relations
:
2770 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2771 nsr_id
, nsd
, r
["provider"], cached_vnfds
, vnf_profile_id
=vnf_profile_id
2773 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2774 nsr_id
, nsd
, r
["requirer"], cached_vnfds
, vnf_profile_id
=vnf_profile_id
2776 provider
= EERelation(relation_provider
)
2777 requirer
= EERelation(relation_requirer
)
2778 relation
= Relation(r
["name"], provider
, requirer
)
2779 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2781 relations
.append(relation
)
2784 def _get_kdu_resource_data(
2786 ee_relation
: EERelation
,
2787 db_nsr
: Dict
[str, Any
],
2788 cached_vnfds
: Dict
[str, Any
],
2789 ) -> DeployedK8sResource
:
2790 nsd
= get_nsd(db_nsr
)
2791 vnf_profiles
= get_vnf_profiles(nsd
)
2792 vnfd_id
= find_in_list(
2794 lambda vnf_profile
: vnf_profile
["id"] == ee_relation
.vnf_profile_id
,
2796 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2797 kdu_resource_profile
= get_kdu_resource_profile(
2798 db_vnfd
, ee_relation
.kdu_resource_profile_id
2800 kdu_name
= kdu_resource_profile
["kdu-name"]
2801 deployed_kdu
, _
= get_deployed_kdu(
2802 db_nsr
.get("_admin", ()).get("deployed", ()),
2804 ee_relation
.vnf_profile_id
,
2806 deployed_kdu
.update({"resource-name": kdu_resource_profile
["resource-name"]})
2809 def _get_deployed_component(
2811 ee_relation
: EERelation
,
2812 db_nsr
: Dict
[str, Any
],
2813 cached_vnfds
: Dict
[str, Any
],
2814 ) -> DeployedComponent
:
2815 nsr_id
= db_nsr
["_id"]
2816 deployed_component
= None
2817 ee_level
= EELevel
.get_level(ee_relation
)
2818 if ee_level
== EELevel
.NS
:
2819 vca
= get_deployed_vca(db_nsr
, {"vdu_id": None, "member-vnf-index": None})
2821 deployed_component
= DeployedVCA(nsr_id
, vca
)
2822 elif ee_level
== EELevel
.VNF
:
2823 vca
= get_deployed_vca(
2827 "member-vnf-index": ee_relation
.vnf_profile_id
,
2828 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2832 deployed_component
= DeployedVCA(nsr_id
, vca
)
2833 elif ee_level
== EELevel
.VDU
:
2834 vca
= get_deployed_vca(
2837 "vdu_id": ee_relation
.vdu_profile_id
,
2838 "member-vnf-index": ee_relation
.vnf_profile_id
,
2839 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2843 deployed_component
= DeployedVCA(nsr_id
, vca
)
2844 elif ee_level
== EELevel
.KDU
:
2845 kdu_resource_data
= self
._get
_kdu
_resource
_data
(
2846 ee_relation
, db_nsr
, cached_vnfds
2848 if kdu_resource_data
:
2849 deployed_component
= DeployedK8sResource(kdu_resource_data
)
2850 return deployed_component
2852 async def _add_relation(
2856 db_nsr
: Dict
[str, Any
],
2857 cached_vnfds
: Dict
[str, Any
],
2858 cached_vnfrs
: Dict
[str, Any
],
2860 deployed_provider
= self
._get
_deployed
_component
(
2861 relation
.provider
, db_nsr
, cached_vnfds
2863 deployed_requirer
= self
._get
_deployed
_component
(
2864 relation
.requirer
, db_nsr
, cached_vnfds
2868 and deployed_requirer
2869 and deployed_provider
.config_sw_installed
2870 and deployed_requirer
.config_sw_installed
2872 provider_db_vnfr
= (
2874 relation
.provider
.nsr_id
,
2875 relation
.provider
.vnf_profile_id
,
2878 if relation
.provider
.vnf_profile_id
2881 requirer_db_vnfr
= (
2883 relation
.requirer
.nsr_id
,
2884 relation
.requirer
.vnf_profile_id
,
2887 if relation
.requirer
.vnf_profile_id
2890 provider_vca_id
= self
.get_vca_id(provider_db_vnfr
, db_nsr
)
2891 requirer_vca_id
= self
.get_vca_id(requirer_db_vnfr
, db_nsr
)
2892 provider_relation_endpoint
= RelationEndpoint(
2893 deployed_provider
.ee_id
,
2895 relation
.provider
.endpoint
,
2897 requirer_relation_endpoint
= RelationEndpoint(
2898 deployed_requirer
.ee_id
,
2900 relation
.requirer
.endpoint
,
2902 await self
.vca_map
[vca_type
].add_relation(
2903 provider
=provider_relation_endpoint
,
2904 requirer
=requirer_relation_endpoint
,
2906 # remove entry from relations list
2910 async def _add_vca_relations(
2916 timeout
: int = 3600,
2920 # 1. find all relations for this VCA
2921 # 2. wait for other peers related
2925 # STEP 1: find all relations for this VCA
2928 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2929 nsd
= get_nsd(db_nsr
)
2932 deployed_vca_dict
= get_deployed_vca_list(db_nsr
)[vca_index
]
2933 my_vca
= DeployedVCA(nsr_id
, deployed_vca_dict
)
2938 relations
.extend(self
._get
_ns
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
2939 relations
.extend(self
._get
_vnf
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
2941 # if no relations, terminate
2943 self
.logger
.debug(logging_text
+ " No relations")
2946 self
.logger
.debug(logging_text
+ " adding relations {}".format(relations
))
2953 if now
- start
>= timeout
:
2954 self
.logger
.error(logging_text
+ " : timeout adding relations")
2957 # reload nsr from database (we need to update record: _admin.deployed.VCA)
2958 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2960 # for each relation, find the VCA's related
2961 for relation
in relations
.copy():
2962 added
= await self
._add
_relation
(
2970 relations
.remove(relation
)
2973 self
.logger
.debug("Relations added")
2975 await asyncio
.sleep(5.0)
2979 except Exception as e
:
2980 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
2983 async def _install_kdu(
2991 k8s_instance_info
: dict,
2992 k8params
: dict = None,
2998 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
3001 "collection": "nsrs",
3002 "filter": {"_id": nsr_id
},
3003 "path": nsr_db_path
,
3006 if k8s_instance_info
.get("kdu-deployment-name"):
3007 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
3009 kdu_instance
= self
.k8scluster_map
[
3011 ].generate_kdu_instance_name(
3012 db_dict
=db_dict_install
,
3013 kdu_model
=k8s_instance_info
["kdu-model"],
3014 kdu_name
=k8s_instance_info
["kdu-name"],
3017 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3019 await self
.k8scluster_map
[k8sclustertype
].install(
3020 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3021 kdu_model
=k8s_instance_info
["kdu-model"],
3024 db_dict
=db_dict_install
,
3026 kdu_name
=k8s_instance_info
["kdu-name"],
3027 namespace
=k8s_instance_info
["namespace"],
3028 kdu_instance
=kdu_instance
,
3032 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3035 # Obtain services to obtain management service ip
3036 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
3037 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3038 kdu_instance
=kdu_instance
,
3039 namespace
=k8s_instance_info
["namespace"],
3042 # Obtain management service info (if exists)
3043 vnfr_update_dict
= {}
3044 kdu_config
= get_configuration(vnfd
, kdud
["name"])
3046 target_ee_list
= kdu_config
.get("execution-environment-list", [])
3051 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
3054 for service
in kdud
.get("service", [])
3055 if service
.get("mgmt-service")
3057 for mgmt_service
in mgmt_services
:
3058 for service
in services
:
3059 if service
["name"].startswith(mgmt_service
["name"]):
3060 # Mgmt service found, Obtain service ip
3061 ip
= service
.get("external_ip", service
.get("cluster_ip"))
3062 if isinstance(ip
, list) and len(ip
) == 1:
3066 "kdur.{}.ip-address".format(kdu_index
)
3069 # Check if must update also mgmt ip at the vnf
3070 service_external_cp
= mgmt_service
.get(
3071 "external-connection-point-ref"
3073 if service_external_cp
:
3075 deep_get(vnfd
, ("mgmt-interface", "cp"))
3076 == service_external_cp
3078 vnfr_update_dict
["ip-address"] = ip
3083 "external-connection-point-ref", ""
3085 == service_external_cp
,
3088 "kdur.{}.ip-address".format(kdu_index
)
3093 "Mgmt service name: {} not found".format(
3094 mgmt_service
["name"]
3098 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3099 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3101 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3104 and kdu_config
.get("initial-config-primitive")
3105 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3107 initial_config_primitive_list
= kdu_config
.get(
3108 "initial-config-primitive"
3110 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3112 for initial_config_primitive
in initial_config_primitive_list
:
3113 primitive_params_
= self
._map
_primitive
_params
(
3114 initial_config_primitive
, {}, {}
3117 await asyncio
.wait_for(
3118 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3119 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3120 kdu_instance
=kdu_instance
,
3121 primitive_name
=initial_config_primitive
["name"],
3122 params
=primitive_params_
,
3123 db_dict
=db_dict_install
,
3129 except Exception as e
:
3130 # Prepare update db with error and raise exception
3133 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3137 vnfr_data
.get("_id"),
3138 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3141 # ignore to keep original exception
3143 # reraise original error
3148 async def deploy_kdus(
3155 task_instantiation_info
,
3157 # Launch kdus if present in the descriptor
3159 k8scluster_id_2_uuic
= {
3160 "helm-chart-v3": {},
3165 async def _get_cluster_id(cluster_id
, cluster_type
):
3166 nonlocal k8scluster_id_2_uuic
3167 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3168 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3170 # check if K8scluster is creating and wait look if previous tasks in process
3171 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3172 "k8scluster", cluster_id
3175 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3176 task_name
, cluster_id
3178 self
.logger
.debug(logging_text
+ text
)
3179 await asyncio
.wait(task_dependency
, timeout
=3600)
3181 db_k8scluster
= self
.db
.get_one(
3182 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3184 if not db_k8scluster
:
3185 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3187 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3189 if cluster_type
== "helm-chart-v3":
3191 # backward compatibility for existing clusters that have not been initialized for helm v3
3192 k8s_credentials
= yaml
.safe_dump(
3193 db_k8scluster
.get("credentials")
3195 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3196 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3198 db_k8scluster_update
= {}
3199 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3200 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3201 db_k8scluster_update
[
3202 "_admin.helm-chart-v3.created"
3204 db_k8scluster_update
[
3205 "_admin.helm-chart-v3.operationalState"
3208 "k8sclusters", cluster_id
, db_k8scluster_update
3210 except Exception as e
:
3213 + "error initializing helm-v3 cluster: {}".format(str(e
))
3216 "K8s cluster '{}' has not been initialized for '{}'".format(
3217 cluster_id
, cluster_type
3222 "K8s cluster '{}' has not been initialized for '{}'".format(
3223 cluster_id
, cluster_type
3226 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3229 logging_text
+= "Deploy kdus: "
3232 db_nsr_update
= {"_admin.deployed.K8s": []}
3233 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3236 updated_cluster_list
= []
3237 updated_v3_cluster_list
= []
3239 for vnfr_data
in db_vnfrs
.values():
3240 vca_id
= self
.get_vca_id(vnfr_data
, {})
3241 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3242 # Step 0: Prepare and set parameters
3243 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3244 vnfd_id
= vnfr_data
.get("vnfd-id")
3245 vnfd_with_id
= find_in_list(
3246 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3250 for kdud
in vnfd_with_id
["kdu"]
3251 if kdud
["name"] == kdur
["kdu-name"]
3253 namespace
= kdur
.get("k8s-namespace")
3254 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3255 if kdur
.get("helm-chart"):
3256 kdumodel
= kdur
["helm-chart"]
3257 # Default version: helm3, if helm-version is v2 assign v2
3258 k8sclustertype
= "helm-chart-v3"
3259 self
.logger
.debug("kdur: {}".format(kdur
))
3261 kdur
.get("helm-version")
3262 and kdur
.get("helm-version") == "v2"
3264 k8sclustertype
= "helm-chart"
3265 elif kdur
.get("juju-bundle"):
3266 kdumodel
= kdur
["juju-bundle"]
3267 k8sclustertype
= "juju-bundle"
3270 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3271 "juju-bundle. Maybe an old NBI version is running".format(
3272 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3275 # check if kdumodel is a file and exists
3277 vnfd_with_id
= find_in_list(
3278 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3280 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3281 if storage
: # may be not present if vnfd has not artifacts
3282 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3283 if storage
["pkg-dir"]:
3284 filename
= "{}/{}/{}s/{}".format(
3291 filename
= "{}/Scripts/{}s/{}".format(
3296 if self
.fs
.file_exists(
3297 filename
, mode
="file"
3298 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3299 kdumodel
= self
.fs
.path
+ filename
3300 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3302 except Exception: # it is not a file
3305 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3306 step
= "Synchronize repos for k8s cluster '{}'".format(
3309 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3313 k8sclustertype
== "helm-chart"
3314 and cluster_uuid
not in updated_cluster_list
3316 k8sclustertype
== "helm-chart-v3"
3317 and cluster_uuid
not in updated_v3_cluster_list
3319 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3320 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3321 cluster_uuid
=cluster_uuid
3324 if del_repo_list
or added_repo_dict
:
3325 if k8sclustertype
== "helm-chart":
3327 "_admin.helm_charts_added." + item
: None
3328 for item
in del_repo_list
3331 "_admin.helm_charts_added." + item
: name
3332 for item
, name
in added_repo_dict
.items()
3334 updated_cluster_list
.append(cluster_uuid
)
3335 elif k8sclustertype
== "helm-chart-v3":
3337 "_admin.helm_charts_v3_added." + item
: None
3338 for item
in del_repo_list
3341 "_admin.helm_charts_v3_added." + item
: name
3342 for item
, name
in added_repo_dict
.items()
3344 updated_v3_cluster_list
.append(cluster_uuid
)
3346 logging_text
+ "repos synchronized on k8s cluster "
3347 "'{}' to_delete: {}, to_add: {}".format(
3348 k8s_cluster_id
, del_repo_list
, added_repo_dict
3353 {"_id": k8s_cluster_id
},
3359 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3360 vnfr_data
["member-vnf-index-ref"],
3364 k8s_instance_info
= {
3365 "kdu-instance": None,
3366 "k8scluster-uuid": cluster_uuid
,
3367 "k8scluster-type": k8sclustertype
,
3368 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3369 "kdu-name": kdur
["kdu-name"],
3370 "kdu-model": kdumodel
,
3371 "namespace": namespace
,
3372 "kdu-deployment-name": kdu_deployment_name
,
3374 db_path
= "_admin.deployed.K8s.{}".format(index
)
3375 db_nsr_update
[db_path
] = k8s_instance_info
3376 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3377 vnfd_with_id
= find_in_list(
3378 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3380 task
= asyncio
.ensure_future(
3389 k8params
=desc_params
,
3394 self
.lcm_tasks
.register(
3398 "instantiate_KDU-{}".format(index
),
3401 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3407 except (LcmException
, asyncio
.CancelledError
):
3409 except Exception as e
:
3410 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3411 if isinstance(e
, (N2VCException
, DbException
)):
3412 self
.logger
.error(logging_text
+ msg
)
3414 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3415 raise LcmException(msg
)
3418 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3437 task_instantiation_info
,
3440 # launch instantiate_N2VC in a asyncio task and register task object
3441 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3442 # if not found, create one entry and update database
3443 # fill db_nsr._admin.deployed.VCA.<index>
3446 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3448 if "execution-environment-list" in descriptor_config
:
3449 ee_list
= descriptor_config
.get("execution-environment-list", [])
3450 elif "juju" in descriptor_config
:
3451 ee_list
= [descriptor_config
] # ns charms
3452 else: # other types as script are not supported
3455 for ee_item
in ee_list
:
3458 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3459 ee_item
.get("juju"), ee_item
.get("helm-chart")
3462 ee_descriptor_id
= ee_item
.get("id")
3463 if ee_item
.get("juju"):
3464 vca_name
= ee_item
["juju"].get("charm")
3467 if ee_item
["juju"].get("charm") is not None
3470 if ee_item
["juju"].get("cloud") == "k8s":
3471 vca_type
= "k8s_proxy_charm"
3472 elif ee_item
["juju"].get("proxy") is False:
3473 vca_type
= "native_charm"
3474 elif ee_item
.get("helm-chart"):
3475 vca_name
= ee_item
["helm-chart"]
3476 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3479 vca_type
= "helm-v3"
3482 logging_text
+ "skipping non juju neither charm configuration"
3487 for vca_index
, vca_deployed
in enumerate(
3488 db_nsr
["_admin"]["deployed"]["VCA"]
3490 if not vca_deployed
:
3493 vca_deployed
.get("member-vnf-index") == member_vnf_index
3494 and vca_deployed
.get("vdu_id") == vdu_id
3495 and vca_deployed
.get("kdu_name") == kdu_name
3496 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3497 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3501 # not found, create one.
3503 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3506 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3508 target
+= "/kdu/{}".format(kdu_name
)
3510 "target_element": target
,
3511 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3512 "member-vnf-index": member_vnf_index
,
3514 "kdu_name": kdu_name
,
3515 "vdu_count_index": vdu_index
,
3516 "operational-status": "init", # TODO revise
3517 "detailed-status": "", # TODO revise
3518 "step": "initial-deploy", # TODO revise
3520 "vdu_name": vdu_name
,
3522 "ee_descriptor_id": ee_descriptor_id
,
3526 # create VCA and configurationStatus in db
3528 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3529 "configurationStatus.{}".format(vca_index
): dict(),
3531 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3533 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3535 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3536 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3537 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3540 task_n2vc
= asyncio
.ensure_future(
3541 self
.instantiate_N2VC(
3542 logging_text
=logging_text
,
3543 vca_index
=vca_index
,
3549 vdu_index
=vdu_index
,
3550 deploy_params
=deploy_params
,
3551 config_descriptor
=descriptor_config
,
3552 base_folder
=base_folder
,
3553 nslcmop_id
=nslcmop_id
,
3557 ee_config_descriptor
=ee_item
,
3560 self
.lcm_tasks
.register(
3564 "instantiate_N2VC-{}".format(vca_index
),
3567 task_instantiation_info
[
3569 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3570 member_vnf_index
or "", vdu_id
or ""
3574 def _create_nslcmop(nsr_id
, operation
, params
):
3576 Creates a ns-lcm-opp content to be stored at database.
3577 :param nsr_id: internal id of the instance
3578 :param operation: instantiate, terminate, scale, action, ...
3579 :param params: user parameters for the operation
3580 :return: dictionary following SOL005 format
3582 # Raise exception if invalid arguments
3583 if not (nsr_id
and operation
and params
):
3585 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3592 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3593 "operationState": "PROCESSING",
3594 "statusEnteredTime": now
,
3595 "nsInstanceId": nsr_id
,
3596 "lcmOperationType": operation
,
3598 "isAutomaticInvocation": False,
3599 "operationParams": params
,
3600 "isCancelPending": False,
3602 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3603 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3608 def _format_additional_params(self
, params
):
3609 params
= params
or {}
3610 for key
, value
in params
.items():
3611 if str(value
).startswith("!!yaml "):
3612 params
[key
] = yaml
.safe_load(value
[7:])
3615 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3616 primitive
= seq
.get("name")
3617 primitive_params
= {}
3619 "member_vnf_index": vnf_index
,
3620 "primitive": primitive
,
3621 "primitive_params": primitive_params
,
3624 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3628 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3629 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3630 if op
.get("operationState") == "COMPLETED":
3631 # b. Skip sub-operation
3632 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3633 return self
.SUBOPERATION_STATUS_SKIP
3635 # c. retry executing sub-operation
3636 # The sub-operation exists, and operationState != 'COMPLETED'
3637 # Update operationState = 'PROCESSING' to indicate a retry.
3638 operationState
= "PROCESSING"
3639 detailed_status
= "In progress"
3640 self
._update
_suboperation
_status
(
3641 db_nslcmop
, op_index
, operationState
, detailed_status
3643 # Return the sub-operation index
3644 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3645 # with arguments extracted from the sub-operation
3648 # Find a sub-operation where all keys in a matching dictionary must match
3649 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3650 def _find_suboperation(self
, db_nslcmop
, match
):
3651 if db_nslcmop
and match
:
3652 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3653 for i
, op
in enumerate(op_list
):
3654 if all(op
.get(k
) == match
[k
] for k
in match
):
3656 return self
.SUBOPERATION_STATUS_NOT_FOUND
3658 # Update status for a sub-operation given its index
3659 def _update_suboperation_status(
3660 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3662 # Update DB for HA tasks
3663 q_filter
= {"_id": db_nslcmop
["_id"]}
3665 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3666 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3669 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3672 # Add sub-operation, return the index of the added sub-operation
3673 # Optionally, set operationState, detailed-status, and operationType
3674 # Status and type are currently set for 'scale' sub-operations:
3675 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3676 # 'detailed-status' : status message
3677 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3678 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3679 def _add_suboperation(
3687 mapped_primitive_params
,
3688 operationState
=None,
3689 detailed_status
=None,
3692 RO_scaling_info
=None,
3695 return self
.SUBOPERATION_STATUS_NOT_FOUND
3696 # Get the "_admin.operations" list, if it exists
3697 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3698 op_list
= db_nslcmop_admin
.get("operations")
3699 # Create or append to the "_admin.operations" list
3701 "member_vnf_index": vnf_index
,
3703 "vdu_count_index": vdu_count_index
,
3704 "primitive": primitive
,
3705 "primitive_params": mapped_primitive_params
,
3708 new_op
["operationState"] = operationState
3710 new_op
["detailed-status"] = detailed_status
3712 new_op
["lcmOperationType"] = operationType
3714 new_op
["RO_nsr_id"] = RO_nsr_id
3716 new_op
["RO_scaling_info"] = RO_scaling_info
3718 # No existing operations, create key 'operations' with current operation as first list element
3719 db_nslcmop_admin
.update({"operations": [new_op
]})
3720 op_list
= db_nslcmop_admin
.get("operations")
3722 # Existing operations, append operation to list
3723 op_list
.append(new_op
)
3725 db_nslcmop_update
= {"_admin.operations": op_list
}
3726 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3727 op_index
= len(op_list
) - 1
3730 # Helper methods for scale() sub-operations
3732 # pre-scale/post-scale:
3733 # Check for 3 different cases:
3734 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3735 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3736 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3737 def _check_or_add_scale_suboperation(
3741 vnf_config_primitive
,
3745 RO_scaling_info
=None,
3747 # Find this sub-operation
3748 if RO_nsr_id
and RO_scaling_info
:
3749 operationType
= "SCALE-RO"
3751 "member_vnf_index": vnf_index
,
3752 "RO_nsr_id": RO_nsr_id
,
3753 "RO_scaling_info": RO_scaling_info
,
3757 "member_vnf_index": vnf_index
,
3758 "primitive": vnf_config_primitive
,
3759 "primitive_params": primitive_params
,
3760 "lcmOperationType": operationType
,
3762 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3763 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3764 # a. New sub-operation
3765 # The sub-operation does not exist, add it.
3766 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3767 # The following parameters are set to None for all kind of scaling:
3769 vdu_count_index
= None
3771 if RO_nsr_id
and RO_scaling_info
:
3772 vnf_config_primitive
= None
3773 primitive_params
= None
3776 RO_scaling_info
= None
3777 # Initial status for sub-operation
3778 operationState
= "PROCESSING"
3779 detailed_status
= "In progress"
3780 # Add sub-operation for pre/post-scaling (zero or more operations)
3781 self
._add
_suboperation
(
3787 vnf_config_primitive
,
3795 return self
.SUBOPERATION_STATUS_NEW
3797 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3798 # or op_index (operationState != 'COMPLETED')
3799 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3801 # Function to return execution_environment id
3803 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3804 # TODO vdu_index_count
3805 for vca
in vca_deployed_list
:
3806 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3809 async def destroy_N2VC(
3817 exec_primitives
=True,
3822 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3823 :param logging_text:
3825 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3826 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3827 :param vca_index: index in the database _admin.deployed.VCA
3828 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3829 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3830 not executed properly
3831 :param scaling_in: True destroys the application, False destroys the model
3832 :return: None or exception
3837 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3838 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3842 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3844 # execute terminate_primitives
3846 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3847 config_descriptor
.get("terminate-config-primitive"),
3848 vca_deployed
.get("ee_descriptor_id"),
3850 vdu_id
= vca_deployed
.get("vdu_id")
3851 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3852 vdu_name
= vca_deployed
.get("vdu_name")
3853 vnf_index
= vca_deployed
.get("member-vnf-index")
3854 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3855 for seq
in terminate_primitives
:
3856 # For each sequence in list, get primitive and call _ns_execute_primitive()
3857 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3858 vnf_index
, seq
.get("name")
3860 self
.logger
.debug(logging_text
+ step
)
3861 # Create the primitive for each sequence, i.e. "primitive": "touch"
3862 primitive
= seq
.get("name")
3863 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3868 self
._add
_suboperation
(
3875 mapped_primitive_params
,
3877 # Sub-operations: Call _ns_execute_primitive() instead of action()
3879 result
, result_detail
= await self
._ns
_execute
_primitive
(
3880 vca_deployed
["ee_id"],
3882 mapped_primitive_params
,
3886 except LcmException
:
3887 # this happens when VCA is not deployed. In this case it is not needed to terminate
3889 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3890 if result
not in result_ok
:
3892 "terminate_primitive {} for vnf_member_index={} fails with "
3893 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3895 # set that this VCA do not need terminated
3896 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3900 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3903 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3904 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3907 await self
.vca_map
[vca_type
].delete_execution_environment(
3908 vca_deployed
["ee_id"],
3909 scaling_in
=scaling_in
,
3914 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3915 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3916 namespace
= "." + db_nsr
["_id"]
3918 await self
.n2vc
.delete_namespace(
3919 namespace
=namespace
,
3920 total_timeout
=self
.timeout_charm_delete
,
3923 except N2VCNotFound
: # already deleted. Skip
3925 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3927 async def _terminate_RO(
3928 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
3931 Terminates a deployment from RO
3932 :param logging_text:
3933 :param nsr_deployed: db_nsr._admin.deployed
3936 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3937 this method will update only the index 2, but it will write on database the concatenated content of the list
3942 ro_nsr_id
= ro_delete_action
= None
3943 if nsr_deployed
and nsr_deployed
.get("RO"):
3944 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3945 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3948 stage
[2] = "Deleting ns from VIM."
3949 db_nsr_update
["detailed-status"] = " ".join(stage
)
3950 self
._write
_op
_status
(nslcmop_id
, stage
)
3951 self
.logger
.debug(logging_text
+ stage
[2])
3952 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3953 self
._write
_op
_status
(nslcmop_id
, stage
)
3954 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3955 ro_delete_action
= desc
["action_id"]
3957 "_admin.deployed.RO.nsr_delete_action_id"
3958 ] = ro_delete_action
3959 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3960 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3961 if ro_delete_action
:
3962 # wait until NS is deleted from VIM
3963 stage
[2] = "Waiting ns deleted from VIM."
3964 detailed_status_old
= None
3968 + " RO_id={} ro_delete_action={}".format(
3969 ro_nsr_id
, ro_delete_action
3972 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3973 self
._write
_op
_status
(nslcmop_id
, stage
)
3975 delete_timeout
= 20 * 60 # 20 minutes
3976 while delete_timeout
> 0:
3977 desc
= await self
.RO
.show(
3979 item_id_name
=ro_nsr_id
,
3980 extra_item
="action",
3981 extra_item_id
=ro_delete_action
,
3985 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3987 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3988 if ns_status
== "ERROR":
3989 raise ROclient
.ROClientException(ns_status_info
)
3990 elif ns_status
== "BUILD":
3991 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3992 elif ns_status
== "ACTIVE":
3993 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3994 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3999 ), "ROclient.check_action_status returns unknown {}".format(
4002 if stage
[2] != detailed_status_old
:
4003 detailed_status_old
= stage
[2]
4004 db_nsr_update
["detailed-status"] = " ".join(stage
)
4005 self
._write
_op
_status
(nslcmop_id
, stage
)
4006 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4007 await asyncio
.sleep(5, loop
=self
.loop
)
4009 else: # delete_timeout <= 0:
4010 raise ROclient
.ROClientException(
4011 "Timeout waiting ns deleted from VIM"
4014 except Exception as e
:
4015 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4017 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4019 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4020 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4021 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4023 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
4026 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4028 failed_detail
.append("delete conflict: {}".format(e
))
4031 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
4034 failed_detail
.append("delete error: {}".format(e
))
4036 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
4040 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
4041 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
4043 stage
[2] = "Deleting nsd from RO."
4044 db_nsr_update
["detailed-status"] = " ".join(stage
)
4045 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4046 self
._write
_op
_status
(nslcmop_id
, stage
)
4047 await self
.RO
.delete("nsd", ro_nsd_id
)
4049 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
4051 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4052 except Exception as e
:
4054 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4056 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4058 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
4061 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4063 failed_detail
.append(
4064 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
4066 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4068 failed_detail
.append(
4069 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
4071 self
.logger
.error(logging_text
+ failed_detail
[-1])
4073 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
4074 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
4075 if not vnf_deployed
or not vnf_deployed
["id"]:
4078 ro_vnfd_id
= vnf_deployed
["id"]
4081 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
4082 vnf_deployed
["member-vnf-index"], ro_vnfd_id
4084 db_nsr_update
["detailed-status"] = " ".join(stage
)
4085 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4086 self
._write
_op
_status
(nslcmop_id
, stage
)
4087 await self
.RO
.delete("vnfd", ro_vnfd_id
)
4089 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
4091 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
4092 except Exception as e
:
4094 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4097 "_admin.deployed.RO.vnfd.{}.id".format(index
)
4101 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4104 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4106 failed_detail
.append(
4107 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4109 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4111 failed_detail
.append(
4112 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4114 self
.logger
.error(logging_text
+ failed_detail
[-1])
4117 stage
[2] = "Error deleting from VIM"
4119 stage
[2] = "Deleted from VIM"
4120 db_nsr_update
["detailed-status"] = " ".join(stage
)
4121 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4122 self
._write
_op
_status
(nslcmop_id
, stage
)
4125 raise LcmException("; ".join(failed_detail
))
4127 async def terminate(self
, nsr_id
, nslcmop_id
):
4128 # Try to lock HA task here
4129 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4130 if not task_is_locked_by_me
:
4133 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4134 self
.logger
.debug(logging_text
+ "Enter")
4135 timeout_ns_terminate
= self
.timeout_ns_terminate
4138 operation_params
= None
4140 error_list
= [] # annotates all failed error messages
4141 db_nslcmop_update
= {}
4142 autoremove
= False # autoremove after terminated
4143 tasks_dict_info
= {}
4146 "Stage 1/3: Preparing task.",
4147 "Waiting for previous operations to terminate.",
4150 # ^ contains [stage, step, VIM-status]
4152 # wait for any previous tasks in process
4153 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4155 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4156 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4157 operation_params
= db_nslcmop
.get("operationParams") or {}
4158 if operation_params
.get("timeout_ns_terminate"):
4159 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4160 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4161 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4163 db_nsr_update
["operational-status"] = "terminating"
4164 db_nsr_update
["config-status"] = "terminating"
4165 self
._write
_ns
_status
(
4167 ns_state
="TERMINATING",
4168 current_operation
="TERMINATING",
4169 current_operation_id
=nslcmop_id
,
4170 other_update
=db_nsr_update
,
4172 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4173 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4174 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4177 stage
[1] = "Getting vnf descriptors from db."
4178 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4180 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4182 db_vnfds_from_id
= {}
4183 db_vnfds_from_member_index
= {}
4185 for vnfr
in db_vnfrs_list
:
4186 vnfd_id
= vnfr
["vnfd-id"]
4187 if vnfd_id
not in db_vnfds_from_id
:
4188 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4189 db_vnfds_from_id
[vnfd_id
] = vnfd
4190 db_vnfds_from_member_index
[
4191 vnfr
["member-vnf-index-ref"]
4192 ] = db_vnfds_from_id
[vnfd_id
]
4194 # Destroy individual execution environments when there are terminating primitives.
4195 # Rest of EE will be deleted at once
4196 # TODO - check before calling _destroy_N2VC
4197 # if not operation_params.get("skip_terminate_primitives"):#
4198 # or not vca.get("needed_terminate"):
4199 stage
[0] = "Stage 2/3 execute terminating primitives."
4200 self
.logger
.debug(logging_text
+ stage
[0])
4201 stage
[1] = "Looking execution environment that needs terminate."
4202 self
.logger
.debug(logging_text
+ stage
[1])
4204 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4205 config_descriptor
= None
4206 vca_member_vnf_index
= vca
.get("member-vnf-index")
4207 vca_id
= self
.get_vca_id(
4208 db_vnfrs_dict
.get(vca_member_vnf_index
)
4209 if vca_member_vnf_index
4213 if not vca
or not vca
.get("ee_id"):
4215 if not vca
.get("member-vnf-index"):
4217 config_descriptor
= db_nsr
.get("ns-configuration")
4218 elif vca
.get("vdu_id"):
4219 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4220 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4221 elif vca
.get("kdu_name"):
4222 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4223 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4225 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4226 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4227 vca_type
= vca
.get("type")
4228 exec_terminate_primitives
= not operation_params
.get(
4229 "skip_terminate_primitives"
4230 ) and vca
.get("needed_terminate")
4231 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4232 # pending native charms
4234 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4236 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4237 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4238 task
= asyncio
.ensure_future(
4246 exec_terminate_primitives
,
4250 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4252 # wait for pending tasks of terminate primitives
4256 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4258 error_list
= await self
._wait
_for
_tasks
(
4261 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4265 tasks_dict_info
.clear()
4267 return # raise LcmException("; ".join(error_list))
4269 # remove All execution environments at once
4270 stage
[0] = "Stage 3/3 delete all."
4272 if nsr_deployed
.get("VCA"):
4273 stage
[1] = "Deleting all execution environments."
4274 self
.logger
.debug(logging_text
+ stage
[1])
4275 vca_id
= self
.get_vca_id({}, db_nsr
)
4276 task_delete_ee
= asyncio
.ensure_future(
4278 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4279 timeout
=self
.timeout_charm_delete
,
4282 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4283 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4285 # Delete from k8scluster
4286 stage
[1] = "Deleting KDUs."
4287 self
.logger
.debug(logging_text
+ stage
[1])
4288 # print(nsr_deployed)
4289 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4290 if not kdu
or not kdu
.get("kdu-instance"):
4292 kdu_instance
= kdu
.get("kdu-instance")
4293 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4294 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4295 vca_id
= self
.get_vca_id({}, db_nsr
)
4296 task_delete_kdu_instance
= asyncio
.ensure_future(
4297 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4298 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4299 kdu_instance
=kdu_instance
,
4306 + "Unknown k8s deployment type {}".format(
4307 kdu
.get("k8scluster-type")
4312 task_delete_kdu_instance
4313 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4316 stage
[1] = "Deleting ns from VIM."
4318 task_delete_ro
= asyncio
.ensure_future(
4319 self
._terminate
_ng
_ro
(
4320 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4324 task_delete_ro
= asyncio
.ensure_future(
4326 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4329 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4331 # rest of staff will be done at finally
4334 ROclient
.ROClientException
,
4339 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4341 except asyncio
.CancelledError
:
4343 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4345 exc
= "Operation was cancelled"
4346 except Exception as e
:
4347 exc
= traceback
.format_exc()
4348 self
.logger
.critical(
4349 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4354 error_list
.append(str(exc
))
4356 # wait for pending tasks
4358 stage
[1] = "Waiting for terminate pending tasks."
4359 self
.logger
.debug(logging_text
+ stage
[1])
4360 error_list
+= await self
._wait
_for
_tasks
(
4363 timeout_ns_terminate
,
4367 stage
[1] = stage
[2] = ""
4368 except asyncio
.CancelledError
:
4369 error_list
.append("Cancelled")
4370 # TODO cancell all tasks
4371 except Exception as exc
:
4372 error_list
.append(str(exc
))
4373 # update status at database
4375 error_detail
= "; ".join(error_list
)
4376 # self.logger.error(logging_text + error_detail)
4377 error_description_nslcmop
= "{} Detail: {}".format(
4378 stage
[0], error_detail
4380 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4381 nslcmop_id
, stage
[0]
4384 db_nsr_update
["operational-status"] = "failed"
4385 db_nsr_update
["detailed-status"] = (
4386 error_description_nsr
+ " Detail: " + error_detail
4388 db_nslcmop_update
["detailed-status"] = error_detail
4389 nslcmop_operation_state
= "FAILED"
4393 error_description_nsr
= error_description_nslcmop
= None
4394 ns_state
= "NOT_INSTANTIATED"
4395 db_nsr_update
["operational-status"] = "terminated"
4396 db_nsr_update
["detailed-status"] = "Done"
4397 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4398 db_nslcmop_update
["detailed-status"] = "Done"
4399 nslcmop_operation_state
= "COMPLETED"
4402 self
._write
_ns
_status
(
4405 current_operation
="IDLE",
4406 current_operation_id
=None,
4407 error_description
=error_description_nsr
,
4408 error_detail
=error_detail
,
4409 other_update
=db_nsr_update
,
4411 self
._write
_op
_status
(
4414 error_message
=error_description_nslcmop
,
4415 operation_state
=nslcmop_operation_state
,
4416 other_update
=db_nslcmop_update
,
4418 if ns_state
== "NOT_INSTANTIATED":
4422 {"nsr-id-ref": nsr_id
},
4423 {"_admin.nsState": "NOT_INSTANTIATED"},
4425 except DbException
as e
:
4428 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4432 if operation_params
:
4433 autoremove
= operation_params
.get("autoremove", False)
4434 if nslcmop_operation_state
:
4436 await self
.msg
.aiowrite(
4441 "nslcmop_id": nslcmop_id
,
4442 "operationState": nslcmop_operation_state
,
4443 "autoremove": autoremove
,
4447 except Exception as e
:
4449 logging_text
+ "kafka_write notification Exception {}".format(e
)
4452 self
.logger
.debug(logging_text
+ "Exit")
4453 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4455 async def _wait_for_tasks(
4456 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4459 error_detail_list
= []
4461 pending_tasks
= list(created_tasks_info
.keys())
4462 num_tasks
= len(pending_tasks
)
4464 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4465 self
._write
_op
_status
(nslcmop_id
, stage
)
4466 while pending_tasks
:
4468 _timeout
= timeout
+ time_start
- time()
4469 done
, pending_tasks
= await asyncio
.wait(
4470 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4472 num_done
+= len(done
)
4473 if not done
: # Timeout
4474 for task
in pending_tasks
:
4475 new_error
= created_tasks_info
[task
] + ": Timeout"
4476 error_detail_list
.append(new_error
)
4477 error_list
.append(new_error
)
4480 if task
.cancelled():
4483 exc
= task
.exception()
4485 if isinstance(exc
, asyncio
.TimeoutError
):
4487 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4488 error_list
.append(created_tasks_info
[task
])
4489 error_detail_list
.append(new_error
)
4496 ROclient
.ROClientException
,
4502 self
.logger
.error(logging_text
+ new_error
)
4504 exc_traceback
= "".join(
4505 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4509 + created_tasks_info
[task
]
4515 logging_text
+ created_tasks_info
[task
] + ": Done"
4517 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4519 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4520 if nsr_id
: # update also nsr
4525 "errorDescription": "Error at: " + ", ".join(error_list
),
4526 "errorDetail": ". ".join(error_detail_list
),
4529 self
._write
_op
_status
(nslcmop_id
, stage
)
4530 return error_detail_list
4533 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4535 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4536 The default-value is used. If it is between < > it look for a value at instantiation_params
4537 :param primitive_desc: portion of VNFD/NSD that describes primitive
4538 :param params: Params provided by user
4539 :param instantiation_params: Instantiation params provided by user
4540 :return: a dictionary with the calculated params
4542 calculated_params
= {}
4543 for parameter
in primitive_desc
.get("parameter", ()):
4544 param_name
= parameter
["name"]
4545 if param_name
in params
:
4546 calculated_params
[param_name
] = params
[param_name
]
4547 elif "default-value" in parameter
or "value" in parameter
:
4548 if "value" in parameter
:
4549 calculated_params
[param_name
] = parameter
["value"]
4551 calculated_params
[param_name
] = parameter
["default-value"]
4553 isinstance(calculated_params
[param_name
], str)
4554 and calculated_params
[param_name
].startswith("<")
4555 and calculated_params
[param_name
].endswith(">")
4557 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4558 calculated_params
[param_name
] = instantiation_params
[
4559 calculated_params
[param_name
][1:-1]
4563 "Parameter {} needed to execute primitive {} not provided".format(
4564 calculated_params
[param_name
], primitive_desc
["name"]
4569 "Parameter {} needed to execute primitive {} not provided".format(
4570 param_name
, primitive_desc
["name"]
4574 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4575 calculated_params
[param_name
] = yaml
.safe_dump(
4576 calculated_params
[param_name
], default_flow_style
=True, width
=256
4578 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4580 ].startswith("!!yaml "):
4581 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4582 if parameter
.get("data-type") == "INTEGER":
4584 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4585 except ValueError: # error converting string to int
4587 "Parameter {} of primitive {} must be integer".format(
4588 param_name
, primitive_desc
["name"]
4591 elif parameter
.get("data-type") == "BOOLEAN":
4592 calculated_params
[param_name
] = not (
4593 (str(calculated_params
[param_name
])).lower() == "false"
4596 # add always ns_config_info if primitive name is config
4597 if primitive_desc
["name"] == "config":
4598 if "ns_config_info" in instantiation_params
:
4599 calculated_params
["ns_config_info"] = instantiation_params
[
4602 return calculated_params
4604 def _look_for_deployed_vca(
4611 ee_descriptor_id
=None,
4613 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4614 for vca
in deployed_vca
:
4617 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4620 vdu_count_index
is not None
4621 and vdu_count_index
!= vca
["vdu_count_index"]
4624 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4626 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4630 # vca_deployed not found
4632 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4633 " is not deployed".format(
4642 ee_id
= vca
.get("ee_id")
4644 "type", "lxc_proxy_charm"
4645 ) # default value for backward compatibility - proxy charm
4648 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4649 "execution environment".format(
4650 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4653 return ee_id
, vca_type
4655 async def _ns_execute_primitive(
4661 retries_interval
=30,
4668 if primitive
== "config":
4669 primitive_params
= {"params": primitive_params
}
4671 vca_type
= vca_type
or "lxc_proxy_charm"
4675 output
= await asyncio
.wait_for(
4676 self
.vca_map
[vca_type
].exec_primitive(
4678 primitive_name
=primitive
,
4679 params_dict
=primitive_params
,
4680 progress_timeout
=self
.timeout_progress_primitive
,
4681 total_timeout
=self
.timeout_primitive
,
4686 timeout
=timeout
or self
.timeout_primitive
,
4690 except asyncio
.CancelledError
:
4692 except Exception as e
: # asyncio.TimeoutError
4693 if isinstance(e
, asyncio
.TimeoutError
):
4698 "Error executing action {} on {} -> {}".format(
4703 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4705 return "FAILED", str(e
)
4707 return "COMPLETED", output
4709 except (LcmException
, asyncio
.CancelledError
):
4711 except Exception as e
:
4712 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4714 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4716 Updating the vca_status with latest juju information in nsrs record
4717 :param: nsr_id: Id of the nsr
4718 :param: nslcmop_id: Id of the nslcmop
4722 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4723 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4724 vca_id
= self
.get_vca_id({}, db_nsr
)
4725 if db_nsr
["_admin"]["deployed"]["K8s"]:
4726 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4727 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4728 await self
._on
_update
_k
8s
_db
(
4729 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4732 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4733 table
, filter = "nsrs", {"_id": nsr_id
}
4734 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4735 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4737 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4738 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4740 async def action(self
, nsr_id
, nslcmop_id
):
4741 # Try to lock HA task here
4742 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4743 if not task_is_locked_by_me
:
4746 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4747 self
.logger
.debug(logging_text
+ "Enter")
4748 # get all needed from database
4752 db_nslcmop_update
= {}
4753 nslcmop_operation_state
= None
4754 error_description_nslcmop
= None
4757 # wait for any previous tasks in process
4758 step
= "Waiting for previous operations to terminate"
4759 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4761 self
._write
_ns
_status
(
4764 current_operation
="RUNNING ACTION",
4765 current_operation_id
=nslcmop_id
,
4768 step
= "Getting information from database"
4769 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4770 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4772 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4773 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4774 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4775 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4776 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4777 primitive
= db_nslcmop
["operationParams"]["primitive"]
4778 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4779 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4780 "timeout_ns_action", self
.timeout_primitive
4784 step
= "Getting vnfr from database"
4785 db_vnfr
= self
.db
.get_one(
4786 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4788 step
= "Getting vnfd from database"
4789 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4791 step
= "Getting nsd from database"
4792 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4794 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4795 # for backward compatibility
4796 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4797 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4798 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4799 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4801 # look for primitive
4802 config_primitive_desc
= descriptor_configuration
= None
4804 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4806 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4808 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4810 descriptor_configuration
= db_nsd
.get("ns-configuration")
4812 if descriptor_configuration
and descriptor_configuration
.get(
4815 for config_primitive
in descriptor_configuration
["config-primitive"]:
4816 if config_primitive
["name"] == primitive
:
4817 config_primitive_desc
= config_primitive
4820 if not config_primitive_desc
:
4821 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4823 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4827 primitive_name
= primitive
4828 ee_descriptor_id
= None
4830 primitive_name
= config_primitive_desc
.get(
4831 "execution-environment-primitive", primitive
4833 ee_descriptor_id
= config_primitive_desc
.get(
4834 "execution-environment-ref"
4840 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4842 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4845 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4847 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4849 desc_params
= parse_yaml_strings(
4850 db_vnfr
.get("additionalParamsForVnf")
4853 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4854 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4855 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4857 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4858 actions
.add(primitive
["name"])
4859 for primitive
in kdu_configuration
.get("config-primitive", []):
4860 actions
.add(primitive
["name"])
4861 kdu_action
= True if primitive_name
in actions
else False
4863 # TODO check if ns is in a proper status
4865 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4867 # kdur and desc_params already set from before
4868 if primitive_params
:
4869 desc_params
.update(primitive_params
)
4870 # TODO Check if we will need something at vnf level
4871 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4873 kdu_name
== kdu
["kdu-name"]
4874 and kdu
["member-vnf-index"] == vnf_index
4879 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4882 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4883 msg
= "unknown k8scluster-type '{}'".format(
4884 kdu
.get("k8scluster-type")
4886 raise LcmException(msg
)
4889 "collection": "nsrs",
4890 "filter": {"_id": nsr_id
},
4891 "path": "_admin.deployed.K8s.{}".format(index
),
4895 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4897 step
= "Executing kdu {}".format(primitive_name
)
4898 if primitive_name
== "upgrade":
4899 if desc_params
.get("kdu_model"):
4900 kdu_model
= desc_params
.get("kdu_model")
4901 del desc_params
["kdu_model"]
4903 kdu_model
= kdu
.get("kdu-model")
4904 parts
= kdu_model
.split(sep
=":")
4906 kdu_model
= parts
[0]
4908 detailed_status
= await asyncio
.wait_for(
4909 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4910 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4911 kdu_instance
=kdu
.get("kdu-instance"),
4913 kdu_model
=kdu_model
,
4916 timeout
=timeout_ns_action
,
4918 timeout
=timeout_ns_action
+ 10,
4921 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4923 elif primitive_name
== "rollback":
4924 detailed_status
= await asyncio
.wait_for(
4925 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4926 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4927 kdu_instance
=kdu
.get("kdu-instance"),
4930 timeout
=timeout_ns_action
,
4932 elif primitive_name
== "status":
4933 detailed_status
= await asyncio
.wait_for(
4934 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4935 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4936 kdu_instance
=kdu
.get("kdu-instance"),
4939 timeout
=timeout_ns_action
,
4942 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
4943 kdu
["kdu-name"], nsr_id
4945 params
= self
._map
_primitive
_params
(
4946 config_primitive_desc
, primitive_params
, desc_params
4949 detailed_status
= await asyncio
.wait_for(
4950 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
4951 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4952 kdu_instance
=kdu_instance
,
4953 primitive_name
=primitive_name
,
4956 timeout
=timeout_ns_action
,
4959 timeout
=timeout_ns_action
,
4963 nslcmop_operation_state
= "COMPLETED"
4965 detailed_status
= ""
4966 nslcmop_operation_state
= "FAILED"
4968 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
4969 nsr_deployed
["VCA"],
4970 member_vnf_index
=vnf_index
,
4972 vdu_count_index
=vdu_count_index
,
4973 ee_descriptor_id
=ee_descriptor_id
,
4975 for vca_index
, vca_deployed
in enumerate(
4976 db_nsr
["_admin"]["deployed"]["VCA"]
4978 if vca_deployed
.get("member-vnf-index") == vnf_index
:
4980 "collection": "nsrs",
4981 "filter": {"_id": nsr_id
},
4982 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
4986 nslcmop_operation_state
,
4988 ) = await self
._ns
_execute
_primitive
(
4990 primitive
=primitive_name
,
4991 primitive_params
=self
._map
_primitive
_params
(
4992 config_primitive_desc
, primitive_params
, desc_params
4994 timeout
=timeout_ns_action
,
5000 db_nslcmop_update
["detailed-status"] = detailed_status
5001 error_description_nslcmop
= (
5002 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
5006 + " task Done with result {} {}".format(
5007 nslcmop_operation_state
, detailed_status
5010 return # database update is called inside finally
5012 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5013 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5015 except asyncio
.CancelledError
:
5017 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5019 exc
= "Operation was cancelled"
5020 except asyncio
.TimeoutError
:
5021 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5023 except Exception as e
:
5024 exc
= traceback
.format_exc()
5025 self
.logger
.critical(
5026 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5035 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5036 nslcmop_operation_state
= "FAILED"
5038 self
._write
_ns
_status
(
5042 ], # TODO check if degraded. For the moment use previous status
5043 current_operation
="IDLE",
5044 current_operation_id
=None,
5045 # error_description=error_description_nsr,
5046 # error_detail=error_detail,
5047 other_update
=db_nsr_update
,
5050 self
._write
_op
_status
(
5053 error_message
=error_description_nslcmop
,
5054 operation_state
=nslcmop_operation_state
,
5055 other_update
=db_nslcmop_update
,
5058 if nslcmop_operation_state
:
5060 await self
.msg
.aiowrite(
5065 "nslcmop_id": nslcmop_id
,
5066 "operationState": nslcmop_operation_state
,
5070 except Exception as e
:
5072 logging_text
+ "kafka_write notification Exception {}".format(e
)
5074 self
.logger
.debug(logging_text
+ "Exit")
5075 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5076 return nslcmop_operation_state
, detailed_status
5078 async def scale(self
, nsr_id
, nslcmop_id
):
5079 # Try to lock HA task here
5080 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5081 if not task_is_locked_by_me
:
5084 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5085 stage
= ["", "", ""]
5086 tasks_dict_info
= {}
5087 # ^ stage, step, VIM progress
5088 self
.logger
.debug(logging_text
+ "Enter")
5089 # get all needed from database
5091 db_nslcmop_update
= {}
5094 # in case of error, indicates what part of scale was failed to put nsr at error status
5095 scale_process
= None
5096 old_operational_status
= ""
5097 old_config_status
= ""
5100 # wait for any previous tasks in process
5101 step
= "Waiting for previous operations to terminate"
5102 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5103 self
._write
_ns
_status
(
5106 current_operation
="SCALING",
5107 current_operation_id
=nslcmop_id
,
5110 step
= "Getting nslcmop from database"
5112 step
+ " after having waited for previous tasks to be completed"
5114 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5116 step
= "Getting nsr from database"
5117 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5118 old_operational_status
= db_nsr
["operational-status"]
5119 old_config_status
= db_nsr
["config-status"]
5121 step
= "Parsing scaling parameters"
5122 db_nsr_update
["operational-status"] = "scaling"
5123 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5124 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5126 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5128 ]["member-vnf-index"]
5129 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5131 ]["scaling-group-descriptor"]
5132 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5133 # for backward compatibility
5134 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5135 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5136 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5137 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5139 step
= "Getting vnfr from database"
5140 db_vnfr
= self
.db
.get_one(
5141 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5144 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5146 step
= "Getting vnfd from database"
5147 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5149 base_folder
= db_vnfd
["_admin"]["storage"]
5151 step
= "Getting scaling-group-descriptor"
5152 scaling_descriptor
= find_in_list(
5153 get_scaling_aspect(db_vnfd
),
5154 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5156 if not scaling_descriptor
:
5158 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5159 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5162 step
= "Sending scale order to VIM"
5163 # TODO check if ns is in a proper status
5165 if not db_nsr
["_admin"].get("scaling-group"):
5170 "_admin.scaling-group": [
5171 {"name": scaling_group
, "nb-scale-op": 0}
5175 admin_scale_index
= 0
5177 for admin_scale_index
, admin_scale_info
in enumerate(
5178 db_nsr
["_admin"]["scaling-group"]
5180 if admin_scale_info
["name"] == scaling_group
:
5181 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5183 else: # not found, set index one plus last element and add new entry with the name
5184 admin_scale_index
+= 1
5186 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5189 vca_scaling_info
= []
5190 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5191 if scaling_type
== "SCALE_OUT":
5192 if "aspect-delta-details" not in scaling_descriptor
:
5194 "Aspect delta details not fount in scaling descriptor {}".format(
5195 scaling_descriptor
["name"]
5198 # count if max-instance-count is reached
5199 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5201 scaling_info
["scaling_direction"] = "OUT"
5202 scaling_info
["vdu-create"] = {}
5203 scaling_info
["kdu-create"] = {}
5204 for delta
in deltas
:
5205 for vdu_delta
in delta
.get("vdu-delta", {}):
5206 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5207 # vdu_index also provides the number of instance of the targeted vdu
5208 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5209 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5213 additional_params
= (
5214 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5217 cloud_init_list
= []
5219 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5220 max_instance_count
= 10
5221 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5222 max_instance_count
= vdu_profile
.get(
5223 "max-number-of-instances", 10
5226 default_instance_num
= get_number_of_instances(
5229 instances_number
= vdu_delta
.get("number-of-instances", 1)
5230 nb_scale_op
+= instances_number
5232 new_instance_count
= nb_scale_op
+ default_instance_num
5233 # Control if new count is over max and vdu count is less than max.
5234 # Then assign new instance count
5235 if new_instance_count
> max_instance_count
> vdu_count
:
5236 instances_number
= new_instance_count
- max_instance_count
5238 instances_number
= instances_number
5240 if new_instance_count
> max_instance_count
:
5242 "reached the limit of {} (max-instance-count) "
5243 "scaling-out operations for the "
5244 "scaling-group-descriptor '{}'".format(
5245 nb_scale_op
, scaling_group
5248 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5250 # TODO Information of its own ip is not available because db_vnfr is not updated.
5251 additional_params
["OSM"] = get_osm_params(
5252 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5254 cloud_init_list
.append(
5255 self
._parse
_cloud
_init
(
5262 vca_scaling_info
.append(
5264 "osm_vdu_id": vdu_delta
["id"],
5265 "member-vnf-index": vnf_index
,
5267 "vdu_index": vdu_index
+ x
,
5270 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5271 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5272 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5273 kdu_name
= kdu_profile
["kdu-name"]
5274 resource_name
= kdu_profile
["resource-name"]
5276 # Might have different kdus in the same delta
5277 # Should have list for each kdu
5278 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5279 scaling_info
["kdu-create"][kdu_name
] = []
5281 kdur
= get_kdur(db_vnfr
, kdu_name
)
5282 if kdur
.get("helm-chart"):
5283 k8s_cluster_type
= "helm-chart-v3"
5284 self
.logger
.debug("kdur: {}".format(kdur
))
5286 kdur
.get("helm-version")
5287 and kdur
.get("helm-version") == "v2"
5289 k8s_cluster_type
= "helm-chart"
5290 raise NotImplementedError
5291 elif kdur
.get("juju-bundle"):
5292 k8s_cluster_type
= "juju-bundle"
5295 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5296 "juju-bundle. Maybe an old NBI version is running".format(
5297 db_vnfr
["member-vnf-index-ref"], kdu_name
5301 max_instance_count
= 10
5302 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5303 max_instance_count
= kdu_profile
.get(
5304 "max-number-of-instances", 10
5307 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5308 deployed_kdu
, _
= get_deployed_kdu(
5309 nsr_deployed
, kdu_name
, vnf_index
5311 if deployed_kdu
is None:
5313 "KDU '{}' for vnf '{}' not deployed".format(
5317 kdu_instance
= deployed_kdu
.get("kdu-instance")
5318 instance_num
= await self
.k8scluster_map
[
5320 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5321 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5322 "number-of-instances", 1
5325 # Control if new count is over max and instance_num is less than max.
5326 # Then assign max instance number to kdu replica count
5327 if kdu_replica_count
> max_instance_count
> instance_num
:
5328 kdu_replica_count
= max_instance_count
5329 if kdu_replica_count
> max_instance_count
:
5331 "reached the limit of {} (max-instance-count) "
5332 "scaling-out operations for the "
5333 "scaling-group-descriptor '{}'".format(
5334 instance_num
, scaling_group
5338 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5339 vca_scaling_info
.append(
5341 "osm_kdu_id": kdu_name
,
5342 "member-vnf-index": vnf_index
,
5344 "kdu_index": instance_num
+ x
- 1,
5347 scaling_info
["kdu-create"][kdu_name
].append(
5349 "member-vnf-index": vnf_index
,
5351 "k8s-cluster-type": k8s_cluster_type
,
5352 "resource-name": resource_name
,
5353 "scale": kdu_replica_count
,
5356 elif scaling_type
== "SCALE_IN":
5357 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5359 scaling_info
["scaling_direction"] = "IN"
5360 scaling_info
["vdu-delete"] = {}
5361 scaling_info
["kdu-delete"] = {}
5363 for delta
in deltas
:
5364 for vdu_delta
in delta
.get("vdu-delta", {}):
5365 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5366 min_instance_count
= 0
5367 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5368 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5369 min_instance_count
= vdu_profile
["min-number-of-instances"]
5371 default_instance_num
= get_number_of_instances(
5372 db_vnfd
, vdu_delta
["id"]
5374 instance_num
= vdu_delta
.get("number-of-instances", 1)
5375 nb_scale_op
-= instance_num
5377 new_instance_count
= nb_scale_op
+ default_instance_num
5379 if new_instance_count
< min_instance_count
< vdu_count
:
5380 instances_number
= min_instance_count
- new_instance_count
5382 instances_number
= instance_num
5384 if new_instance_count
< min_instance_count
:
5386 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5387 "scaling-group-descriptor '{}'".format(
5388 nb_scale_op
, scaling_group
5391 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5392 vca_scaling_info
.append(
5394 "osm_vdu_id": vdu_delta
["id"],
5395 "member-vnf-index": vnf_index
,
5397 "vdu_index": vdu_index
- 1 - x
,
5400 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5401 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5402 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5403 kdu_name
= kdu_profile
["kdu-name"]
5404 resource_name
= kdu_profile
["resource-name"]
5406 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5407 scaling_info
["kdu-delete"][kdu_name
] = []
5409 kdur
= get_kdur(db_vnfr
, kdu_name
)
5410 if kdur
.get("helm-chart"):
5411 k8s_cluster_type
= "helm-chart-v3"
5412 self
.logger
.debug("kdur: {}".format(kdur
))
5414 kdur
.get("helm-version")
5415 and kdur
.get("helm-version") == "v2"
5417 k8s_cluster_type
= "helm-chart"
5418 raise NotImplementedError
5419 elif kdur
.get("juju-bundle"):
5420 k8s_cluster_type
= "juju-bundle"
5423 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5424 "juju-bundle. Maybe an old NBI version is running".format(
5425 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5429 min_instance_count
= 0
5430 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5431 min_instance_count
= kdu_profile
["min-number-of-instances"]
5433 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5434 deployed_kdu
, _
= get_deployed_kdu(
5435 nsr_deployed
, kdu_name
, vnf_index
5437 if deployed_kdu
is None:
5439 "KDU '{}' for vnf '{}' not deployed".format(
5443 kdu_instance
= deployed_kdu
.get("kdu-instance")
5444 instance_num
= await self
.k8scluster_map
[
5446 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5447 kdu_replica_count
= instance_num
- kdu_delta
.get(
5448 "number-of-instances", 1
5451 if kdu_replica_count
< min_instance_count
< instance_num
:
5452 kdu_replica_count
= min_instance_count
5453 if kdu_replica_count
< min_instance_count
:
5455 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5456 "scaling-group-descriptor '{}'".format(
5457 instance_num
, scaling_group
5461 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5462 vca_scaling_info
.append(
5464 "osm_kdu_id": kdu_name
,
5465 "member-vnf-index": vnf_index
,
5467 "kdu_index": instance_num
- x
- 1,
5470 scaling_info
["kdu-delete"][kdu_name
].append(
5472 "member-vnf-index": vnf_index
,
5474 "k8s-cluster-type": k8s_cluster_type
,
5475 "resource-name": resource_name
,
5476 "scale": kdu_replica_count
,
5480 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5481 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5482 if scaling_info
["scaling_direction"] == "IN":
5483 for vdur
in reversed(db_vnfr
["vdur"]):
5484 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5485 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5486 scaling_info
["vdu"].append(
5488 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5489 "vdu_id": vdur
["vdu-id-ref"],
5493 for interface
in vdur
["interfaces"]:
5494 scaling_info
["vdu"][-1]["interface"].append(
5496 "name": interface
["name"],
5497 "ip_address": interface
["ip-address"],
5498 "mac_address": interface
.get("mac-address"),
5501 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5504 step
= "Executing pre-scale vnf-config-primitive"
5505 if scaling_descriptor
.get("scaling-config-action"):
5506 for scaling_config_action
in scaling_descriptor
[
5507 "scaling-config-action"
5510 scaling_config_action
.get("trigger") == "pre-scale-in"
5511 and scaling_type
== "SCALE_IN"
5513 scaling_config_action
.get("trigger") == "pre-scale-out"
5514 and scaling_type
== "SCALE_OUT"
5516 vnf_config_primitive
= scaling_config_action
[
5517 "vnf-config-primitive-name-ref"
5519 step
= db_nslcmop_update
[
5521 ] = "executing pre-scale scaling-config-action '{}'".format(
5522 vnf_config_primitive
5525 # look for primitive
5526 for config_primitive
in (
5527 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5528 ).get("config-primitive", ()):
5529 if config_primitive
["name"] == vnf_config_primitive
:
5533 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5534 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5535 "primitive".format(scaling_group
, vnf_config_primitive
)
5538 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5539 if db_vnfr
.get("additionalParamsForVnf"):
5540 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5542 scale_process
= "VCA"
5543 db_nsr_update
["config-status"] = "configuring pre-scaling"
5544 primitive_params
= self
._map
_primitive
_params
(
5545 config_primitive
, {}, vnfr_params
5548 # Pre-scale retry check: Check if this sub-operation has been executed before
5549 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5552 vnf_config_primitive
,
5556 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5557 # Skip sub-operation
5558 result
= "COMPLETED"
5559 result_detail
= "Done"
5562 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5563 vnf_config_primitive
, result
, result_detail
5567 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5568 # New sub-operation: Get index of this sub-operation
5570 len(db_nslcmop
.get("_admin", {}).get("operations"))
5575 + "vnf_config_primitive={} New sub-operation".format(
5576 vnf_config_primitive
5580 # retry: Get registered params for this existing sub-operation
5581 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5584 vnf_index
= op
.get("member_vnf_index")
5585 vnf_config_primitive
= op
.get("primitive")
5586 primitive_params
= op
.get("primitive_params")
5589 + "vnf_config_primitive={} Sub-operation retry".format(
5590 vnf_config_primitive
5593 # Execute the primitive, either with new (first-time) or registered (reintent) args
5594 ee_descriptor_id
= config_primitive
.get(
5595 "execution-environment-ref"
5597 primitive_name
= config_primitive
.get(
5598 "execution-environment-primitive", vnf_config_primitive
5600 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5601 nsr_deployed
["VCA"],
5602 member_vnf_index
=vnf_index
,
5604 vdu_count_index
=None,
5605 ee_descriptor_id
=ee_descriptor_id
,
5607 result
, result_detail
= await self
._ns
_execute
_primitive
(
5616 + "vnf_config_primitive={} Done with result {} {}".format(
5617 vnf_config_primitive
, result
, result_detail
5620 # Update operationState = COMPLETED | FAILED
5621 self
._update
_suboperation
_status
(
5622 db_nslcmop
, op_index
, result
, result_detail
5625 if result
== "FAILED":
5626 raise LcmException(result_detail
)
5627 db_nsr_update
["config-status"] = old_config_status
5628 scale_process
= None
5632 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5635 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5638 # SCALE-IN VCA - BEGIN
5639 if vca_scaling_info
:
5640 step
= db_nslcmop_update
[
5642 ] = "Deleting the execution environments"
5643 scale_process
= "VCA"
5644 for vca_info
in vca_scaling_info
:
5645 if vca_info
["type"] == "delete":
5646 member_vnf_index
= str(vca_info
["member-vnf-index"])
5648 logging_text
+ "vdu info: {}".format(vca_info
)
5650 if vca_info
.get("osm_vdu_id"):
5651 vdu_id
= vca_info
["osm_vdu_id"]
5652 vdu_index
= int(vca_info
["vdu_index"])
5655 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5656 member_vnf_index
, vdu_id
, vdu_index
5660 kdu_id
= vca_info
["osm_kdu_id"]
5663 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5664 member_vnf_index
, kdu_id
, vdu_index
5666 stage
[2] = step
= "Scaling in VCA"
5667 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5668 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5669 config_update
= db_nsr
["configurationStatus"]
5670 for vca_index
, vca
in enumerate(vca_update
):
5672 (vca
or vca
.get("ee_id"))
5673 and vca
["member-vnf-index"] == member_vnf_index
5674 and vca
["vdu_count_index"] == vdu_index
5676 if vca
.get("vdu_id"):
5677 config_descriptor
= get_configuration(
5678 db_vnfd
, vca
.get("vdu_id")
5680 elif vca
.get("kdu_name"):
5681 config_descriptor
= get_configuration(
5682 db_vnfd
, vca
.get("kdu_name")
5685 config_descriptor
= get_configuration(
5686 db_vnfd
, db_vnfd
["id"]
5688 operation_params
= (
5689 db_nslcmop
.get("operationParams") or {}
5691 exec_terminate_primitives
= not operation_params
.get(
5692 "skip_terminate_primitives"
5693 ) and vca
.get("needed_terminate")
5694 task
= asyncio
.ensure_future(
5703 exec_primitives
=exec_terminate_primitives
,
5707 timeout
=self
.timeout_charm_delete
,
5710 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5713 del vca_update
[vca_index
]
5714 del config_update
[vca_index
]
5715 # wait for pending tasks of terminate primitives
5719 + "Waiting for tasks {}".format(
5720 list(tasks_dict_info
.keys())
5723 error_list
= await self
._wait
_for
_tasks
(
5727 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5732 tasks_dict_info
.clear()
5734 raise LcmException("; ".join(error_list
))
5736 db_vca_and_config_update
= {
5737 "_admin.deployed.VCA": vca_update
,
5738 "configurationStatus": config_update
,
5741 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5743 scale_process
= None
5744 # SCALE-IN VCA - END
5747 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5748 scale_process
= "RO"
5749 if self
.ro_config
.get("ng"):
5750 await self
._scale
_ng
_ro
(
5751 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5753 scaling_info
.pop("vdu-create", None)
5754 scaling_info
.pop("vdu-delete", None)
5756 scale_process
= None
5760 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5761 scale_process
= "KDU"
5762 await self
._scale
_kdu
(
5763 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5765 scaling_info
.pop("kdu-create", None)
5766 scaling_info
.pop("kdu-delete", None)
5768 scale_process
= None
5772 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5774 # SCALE-UP VCA - BEGIN
5775 if vca_scaling_info
:
5776 step
= db_nslcmop_update
[
5778 ] = "Creating new execution environments"
5779 scale_process
= "VCA"
5780 for vca_info
in vca_scaling_info
:
5781 if vca_info
["type"] == "create":
5782 member_vnf_index
= str(vca_info
["member-vnf-index"])
5784 logging_text
+ "vdu info: {}".format(vca_info
)
5786 vnfd_id
= db_vnfr
["vnfd-ref"]
5787 if vca_info
.get("osm_vdu_id"):
5788 vdu_index
= int(vca_info
["vdu_index"])
5789 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5790 if db_vnfr
.get("additionalParamsForVnf"):
5791 deploy_params
.update(
5793 db_vnfr
["additionalParamsForVnf"].copy()
5796 descriptor_config
= get_configuration(
5797 db_vnfd
, db_vnfd
["id"]
5799 if descriptor_config
:
5804 logging_text
=logging_text
5805 + "member_vnf_index={} ".format(member_vnf_index
),
5808 nslcmop_id
=nslcmop_id
,
5814 member_vnf_index
=member_vnf_index
,
5815 vdu_index
=vdu_index
,
5817 deploy_params
=deploy_params
,
5818 descriptor_config
=descriptor_config
,
5819 base_folder
=base_folder
,
5820 task_instantiation_info
=tasks_dict_info
,
5823 vdu_id
= vca_info
["osm_vdu_id"]
5824 vdur
= find_in_list(
5825 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5827 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5828 if vdur
.get("additionalParams"):
5829 deploy_params_vdu
= parse_yaml_strings(
5830 vdur
["additionalParams"]
5833 deploy_params_vdu
= deploy_params
5834 deploy_params_vdu
["OSM"] = get_osm_params(
5835 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5837 if descriptor_config
:
5842 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5843 member_vnf_index
, vdu_id
, vdu_index
5845 stage
[2] = step
= "Scaling out VCA"
5846 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5848 logging_text
=logging_text
5849 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5850 member_vnf_index
, vdu_id
, vdu_index
5854 nslcmop_id
=nslcmop_id
,
5860 member_vnf_index
=member_vnf_index
,
5861 vdu_index
=vdu_index
,
5863 deploy_params
=deploy_params_vdu
,
5864 descriptor_config
=descriptor_config
,
5865 base_folder
=base_folder
,
5866 task_instantiation_info
=tasks_dict_info
,
5870 kdu_name
= vca_info
["osm_kdu_id"]
5871 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5872 if descriptor_config
:
5874 kdu_index
= int(vca_info
["kdu_index"])
5878 for x
in db_vnfr
["kdur"]
5879 if x
["kdu-name"] == kdu_name
5881 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5882 if kdur
.get("additionalParams"):
5883 deploy_params_kdu
= parse_yaml_strings(
5884 kdur
["additionalParams"]
5888 logging_text
=logging_text
,
5891 nslcmop_id
=nslcmop_id
,
5897 member_vnf_index
=member_vnf_index
,
5898 vdu_index
=kdu_index
,
5900 deploy_params
=deploy_params_kdu
,
5901 descriptor_config
=descriptor_config
,
5902 base_folder
=base_folder
,
5903 task_instantiation_info
=tasks_dict_info
,
5906 # SCALE-UP VCA - END
5907 scale_process
= None
5910 # execute primitive service POST-SCALING
5911 step
= "Executing post-scale vnf-config-primitive"
5912 if scaling_descriptor
.get("scaling-config-action"):
5913 for scaling_config_action
in scaling_descriptor
[
5914 "scaling-config-action"
5917 scaling_config_action
.get("trigger") == "post-scale-in"
5918 and scaling_type
== "SCALE_IN"
5920 scaling_config_action
.get("trigger") == "post-scale-out"
5921 and scaling_type
== "SCALE_OUT"
5923 vnf_config_primitive
= scaling_config_action
[
5924 "vnf-config-primitive-name-ref"
5926 step
= db_nslcmop_update
[
5928 ] = "executing post-scale scaling-config-action '{}'".format(
5929 vnf_config_primitive
5932 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5933 if db_vnfr
.get("additionalParamsForVnf"):
5934 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5936 # look for primitive
5937 for config_primitive
in (
5938 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5939 ).get("config-primitive", ()):
5940 if config_primitive
["name"] == vnf_config_primitive
:
5944 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
5945 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
5946 "config-primitive".format(
5947 scaling_group
, vnf_config_primitive
5950 scale_process
= "VCA"
5951 db_nsr_update
["config-status"] = "configuring post-scaling"
5952 primitive_params
= self
._map
_primitive
_params
(
5953 config_primitive
, {}, vnfr_params
5956 # Post-scale retry check: Check if this sub-operation has been executed before
5957 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5960 vnf_config_primitive
,
5964 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5965 # Skip sub-operation
5966 result
= "COMPLETED"
5967 result_detail
= "Done"
5970 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5971 vnf_config_primitive
, result
, result_detail
5975 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5976 # New sub-operation: Get index of this sub-operation
5978 len(db_nslcmop
.get("_admin", {}).get("operations"))
5983 + "vnf_config_primitive={} New sub-operation".format(
5984 vnf_config_primitive
5988 # retry: Get registered params for this existing sub-operation
5989 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5992 vnf_index
= op
.get("member_vnf_index")
5993 vnf_config_primitive
= op
.get("primitive")
5994 primitive_params
= op
.get("primitive_params")
5997 + "vnf_config_primitive={} Sub-operation retry".format(
5998 vnf_config_primitive
6001 # Execute the primitive, either with new (first-time) or registered (reintent) args
6002 ee_descriptor_id
= config_primitive
.get(
6003 "execution-environment-ref"
6005 primitive_name
= config_primitive
.get(
6006 "execution-environment-primitive", vnf_config_primitive
6008 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6009 nsr_deployed
["VCA"],
6010 member_vnf_index
=vnf_index
,
6012 vdu_count_index
=None,
6013 ee_descriptor_id
=ee_descriptor_id
,
6015 result
, result_detail
= await self
._ns
_execute
_primitive
(
6024 + "vnf_config_primitive={} Done with result {} {}".format(
6025 vnf_config_primitive
, result
, result_detail
6028 # Update operationState = COMPLETED | FAILED
6029 self
._update
_suboperation
_status
(
6030 db_nslcmop
, op_index
, result
, result_detail
6033 if result
== "FAILED":
6034 raise LcmException(result_detail
)
6035 db_nsr_update
["config-status"] = old_config_status
6036 scale_process
= None
6041 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6042 db_nsr_update
["operational-status"] = (
6044 if old_operational_status
== "failed"
6045 else old_operational_status
6047 db_nsr_update
["config-status"] = old_config_status
6050 ROclient
.ROClientException
,
6055 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6057 except asyncio
.CancelledError
:
6059 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6061 exc
= "Operation was cancelled"
6062 except Exception as e
:
6063 exc
= traceback
.format_exc()
6064 self
.logger
.critical(
6065 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6069 self
._write
_ns
_status
(
6072 current_operation
="IDLE",
6073 current_operation_id
=None,
6076 stage
[1] = "Waiting for instantiate pending tasks."
6077 self
.logger
.debug(logging_text
+ stage
[1])
6078 exc
= await self
._wait
_for
_tasks
(
6081 self
.timeout_ns_deploy
,
6089 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6090 nslcmop_operation_state
= "FAILED"
6092 db_nsr_update
["operational-status"] = old_operational_status
6093 db_nsr_update
["config-status"] = old_config_status
6094 db_nsr_update
["detailed-status"] = ""
6096 if "VCA" in scale_process
:
6097 db_nsr_update
["config-status"] = "failed"
6098 if "RO" in scale_process
:
6099 db_nsr_update
["operational-status"] = "failed"
6102 ] = "FAILED scaling nslcmop={} {}: {}".format(
6103 nslcmop_id
, step
, exc
6106 error_description_nslcmop
= None
6107 nslcmop_operation_state
= "COMPLETED"
6108 db_nslcmop_update
["detailed-status"] = "Done"
6110 self
._write
_op
_status
(
6113 error_message
=error_description_nslcmop
,
6114 operation_state
=nslcmop_operation_state
,
6115 other_update
=db_nslcmop_update
,
6118 self
._write
_ns
_status
(
6121 current_operation
="IDLE",
6122 current_operation_id
=None,
6123 other_update
=db_nsr_update
,
6126 if nslcmop_operation_state
:
6130 "nslcmop_id": nslcmop_id
,
6131 "operationState": nslcmop_operation_state
,
6133 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6134 except Exception as e
:
6136 logging_text
+ "kafka_write notification Exception {}".format(e
)
6138 self
.logger
.debug(logging_text
+ "Exit")
6139 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6141 async def _scale_kdu(
6142 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6144 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6145 for kdu_name
in _scaling_info
:
6146 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6147 deployed_kdu
, index
= get_deployed_kdu(
6148 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6150 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6151 kdu_instance
= deployed_kdu
["kdu-instance"]
6152 scale
= int(kdu_scaling_info
["scale"])
6153 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6156 "collection": "nsrs",
6157 "filter": {"_id": nsr_id
},
6158 "path": "_admin.deployed.K8s.{}".format(index
),
6161 step
= "scaling application {}".format(
6162 kdu_scaling_info
["resource-name"]
6164 self
.logger
.debug(logging_text
+ step
)
6166 if kdu_scaling_info
["type"] == "delete":
6167 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6170 and kdu_config
.get("terminate-config-primitive")
6171 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6173 terminate_config_primitive_list
= kdu_config
.get(
6174 "terminate-config-primitive"
6176 terminate_config_primitive_list
.sort(
6177 key
=lambda val
: int(val
["seq"])
6181 terminate_config_primitive
6182 ) in terminate_config_primitive_list
:
6183 primitive_params_
= self
._map
_primitive
_params
(
6184 terminate_config_primitive
, {}, {}
6186 step
= "execute terminate config primitive"
6187 self
.logger
.debug(logging_text
+ step
)
6188 await asyncio
.wait_for(
6189 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6190 cluster_uuid
=cluster_uuid
,
6191 kdu_instance
=kdu_instance
,
6192 primitive_name
=terminate_config_primitive
["name"],
6193 params
=primitive_params_
,
6200 await asyncio
.wait_for(
6201 self
.k8scluster_map
[k8s_cluster_type
].scale(
6204 kdu_scaling_info
["resource-name"],
6207 timeout
=self
.timeout_vca_on_error
,
6210 if kdu_scaling_info
["type"] == "create":
6211 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6214 and kdu_config
.get("initial-config-primitive")
6215 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6217 initial_config_primitive_list
= kdu_config
.get(
6218 "initial-config-primitive"
6220 initial_config_primitive_list
.sort(
6221 key
=lambda val
: int(val
["seq"])
6224 for initial_config_primitive
in initial_config_primitive_list
:
6225 primitive_params_
= self
._map
_primitive
_params
(
6226 initial_config_primitive
, {}, {}
6228 step
= "execute initial config primitive"
6229 self
.logger
.debug(logging_text
+ step
)
6230 await asyncio
.wait_for(
6231 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6232 cluster_uuid
=cluster_uuid
,
6233 kdu_instance
=kdu_instance
,
6234 primitive_name
=initial_config_primitive
["name"],
6235 params
=primitive_params_
,
6242 async def _scale_ng_ro(
6243 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6245 nsr_id
= db_nslcmop
["nsInstanceId"]
6246 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6249 # read from db: vnfd's for every vnf
6252 # for each vnf in ns, read vnfd
6253 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6254 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6255 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6256 # if we haven't this vnfd, read it from db
6257 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6259 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6260 db_vnfds
.append(vnfd
)
6261 n2vc_key
= self
.n2vc
.get_public_key()
6262 n2vc_key_list
= [n2vc_key
]
6265 vdu_scaling_info
.get("vdu-create"),
6266 vdu_scaling_info
.get("vdu-delete"),
6269 # db_vnfr has been updated, update db_vnfrs to use it
6270 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6271 await self
._instantiate
_ng
_ro
(
6281 start_deploy
=time(),
6282 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6284 if vdu_scaling_info
.get("vdu-delete"):
6286 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6289 async def add_prometheus_metrics(
6290 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6292 if not self
.prometheus
:
6294 # look if exist a file called 'prometheus*.j2' and
6295 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6299 for f
in artifact_content
6300 if f
.startswith("prometheus") and f
.endswith(".j2")
6306 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6310 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6311 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6313 vnfr_id
= vnfr_id
.replace("-", "")
6315 "JOB_NAME": vnfr_id
,
6316 "TARGET_IP": target_ip
,
6317 "EXPORTER_POD_IP": host_name
,
6318 "EXPORTER_POD_PORT": host_port
,
6320 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
6321 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6322 for job
in job_list
:
6324 not isinstance(job
.get("job_name"), str)
6325 or vnfr_id
not in job
["job_name"]
6327 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6328 job
["nsr_id"] = nsr_id
6329 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
6330 if await self
.prometheus
.update(job_dict
):
6331 return list(job_dict
.keys())
6333 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6335 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6337 :param: vim_account_id: VIM Account ID
6339 :return: (cloud_name, cloud_credential)
6341 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6342 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6344 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6346 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6348 :param: vim_account_id: VIM Account ID
6350 :return: (cloud_name, cloud_credential)
6352 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6353 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")