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 # Sync filesystem before running a primitive
4792 self
.fs
.sync(db_vnfr
["vnfd-id"])
4794 step
= "Getting nsd from database"
4795 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4797 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4798 # for backward compatibility
4799 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4800 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4801 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4802 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4804 # look for primitive
4805 config_primitive_desc
= descriptor_configuration
= None
4807 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4809 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4811 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4813 descriptor_configuration
= db_nsd
.get("ns-configuration")
4815 if descriptor_configuration
and descriptor_configuration
.get(
4818 for config_primitive
in descriptor_configuration
["config-primitive"]:
4819 if config_primitive
["name"] == primitive
:
4820 config_primitive_desc
= config_primitive
4823 if not config_primitive_desc
:
4824 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4826 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4830 primitive_name
= primitive
4831 ee_descriptor_id
= None
4833 primitive_name
= config_primitive_desc
.get(
4834 "execution-environment-primitive", primitive
4836 ee_descriptor_id
= config_primitive_desc
.get(
4837 "execution-environment-ref"
4843 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4845 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4848 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4850 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4852 desc_params
= parse_yaml_strings(
4853 db_vnfr
.get("additionalParamsForVnf")
4856 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4857 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4858 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4860 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4861 actions
.add(primitive
["name"])
4862 for primitive
in kdu_configuration
.get("config-primitive", []):
4863 actions
.add(primitive
["name"])
4864 kdu_action
= True if primitive_name
in actions
else False
4866 # TODO check if ns is in a proper status
4868 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4870 # kdur and desc_params already set from before
4871 if primitive_params
:
4872 desc_params
.update(primitive_params
)
4873 # TODO Check if we will need something at vnf level
4874 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4876 kdu_name
== kdu
["kdu-name"]
4877 and kdu
["member-vnf-index"] == vnf_index
4882 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4885 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4886 msg
= "unknown k8scluster-type '{}'".format(
4887 kdu
.get("k8scluster-type")
4889 raise LcmException(msg
)
4892 "collection": "nsrs",
4893 "filter": {"_id": nsr_id
},
4894 "path": "_admin.deployed.K8s.{}".format(index
),
4898 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4900 step
= "Executing kdu {}".format(primitive_name
)
4901 if primitive_name
== "upgrade":
4902 if desc_params
.get("kdu_model"):
4903 kdu_model
= desc_params
.get("kdu_model")
4904 del desc_params
["kdu_model"]
4906 kdu_model
= kdu
.get("kdu-model")
4907 parts
= kdu_model
.split(sep
=":")
4909 kdu_model
= parts
[0]
4911 detailed_status
= await asyncio
.wait_for(
4912 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4913 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4914 kdu_instance
=kdu
.get("kdu-instance"),
4916 kdu_model
=kdu_model
,
4919 timeout
=timeout_ns_action
,
4921 timeout
=timeout_ns_action
+ 10,
4924 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4926 elif primitive_name
== "rollback":
4927 detailed_status
= await asyncio
.wait_for(
4928 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4929 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4930 kdu_instance
=kdu
.get("kdu-instance"),
4933 timeout
=timeout_ns_action
,
4935 elif primitive_name
== "status":
4936 detailed_status
= await asyncio
.wait_for(
4937 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4938 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4939 kdu_instance
=kdu
.get("kdu-instance"),
4942 timeout
=timeout_ns_action
,
4945 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
4946 kdu
["kdu-name"], nsr_id
4948 params
= self
._map
_primitive
_params
(
4949 config_primitive_desc
, primitive_params
, desc_params
4952 detailed_status
= await asyncio
.wait_for(
4953 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
4954 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4955 kdu_instance
=kdu_instance
,
4956 primitive_name
=primitive_name
,
4959 timeout
=timeout_ns_action
,
4962 timeout
=timeout_ns_action
,
4966 nslcmop_operation_state
= "COMPLETED"
4968 detailed_status
= ""
4969 nslcmop_operation_state
= "FAILED"
4971 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
4972 nsr_deployed
["VCA"],
4973 member_vnf_index
=vnf_index
,
4975 vdu_count_index
=vdu_count_index
,
4976 ee_descriptor_id
=ee_descriptor_id
,
4978 for vca_index
, vca_deployed
in enumerate(
4979 db_nsr
["_admin"]["deployed"]["VCA"]
4981 if vca_deployed
.get("member-vnf-index") == vnf_index
:
4983 "collection": "nsrs",
4984 "filter": {"_id": nsr_id
},
4985 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
4989 nslcmop_operation_state
,
4991 ) = await self
._ns
_execute
_primitive
(
4993 primitive
=primitive_name
,
4994 primitive_params
=self
._map
_primitive
_params
(
4995 config_primitive_desc
, primitive_params
, desc_params
4997 timeout
=timeout_ns_action
,
5003 db_nslcmop_update
["detailed-status"] = detailed_status
5004 error_description_nslcmop
= (
5005 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
5009 + " task Done with result {} {}".format(
5010 nslcmop_operation_state
, detailed_status
5013 return # database update is called inside finally
5015 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5016 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5018 except asyncio
.CancelledError
:
5020 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5022 exc
= "Operation was cancelled"
5023 except asyncio
.TimeoutError
:
5024 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5026 except Exception as e
:
5027 exc
= traceback
.format_exc()
5028 self
.logger
.critical(
5029 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5038 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5039 nslcmop_operation_state
= "FAILED"
5041 self
._write
_ns
_status
(
5045 ], # TODO check if degraded. For the moment use previous status
5046 current_operation
="IDLE",
5047 current_operation_id
=None,
5048 # error_description=error_description_nsr,
5049 # error_detail=error_detail,
5050 other_update
=db_nsr_update
,
5053 self
._write
_op
_status
(
5056 error_message
=error_description_nslcmop
,
5057 operation_state
=nslcmop_operation_state
,
5058 other_update
=db_nslcmop_update
,
5061 if nslcmop_operation_state
:
5063 await self
.msg
.aiowrite(
5068 "nslcmop_id": nslcmop_id
,
5069 "operationState": nslcmop_operation_state
,
5073 except Exception as e
:
5075 logging_text
+ "kafka_write notification Exception {}".format(e
)
5077 self
.logger
.debug(logging_text
+ "Exit")
5078 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5079 return nslcmop_operation_state
, detailed_status
5081 async def scale(self
, nsr_id
, nslcmop_id
):
5082 # Try to lock HA task here
5083 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5084 if not task_is_locked_by_me
:
5087 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5088 stage
= ["", "", ""]
5089 tasks_dict_info
= {}
5090 # ^ stage, step, VIM progress
5091 self
.logger
.debug(logging_text
+ "Enter")
5092 # get all needed from database
5094 db_nslcmop_update
= {}
5097 # in case of error, indicates what part of scale was failed to put nsr at error status
5098 scale_process
= None
5099 old_operational_status
= ""
5100 old_config_status
= ""
5103 # wait for any previous tasks in process
5104 step
= "Waiting for previous operations to terminate"
5105 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5106 self
._write
_ns
_status
(
5109 current_operation
="SCALING",
5110 current_operation_id
=nslcmop_id
,
5113 step
= "Getting nslcmop from database"
5115 step
+ " after having waited for previous tasks to be completed"
5117 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5119 step
= "Getting nsr from database"
5120 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5121 old_operational_status
= db_nsr
["operational-status"]
5122 old_config_status
= db_nsr
["config-status"]
5124 step
= "Parsing scaling parameters"
5125 db_nsr_update
["operational-status"] = "scaling"
5126 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5127 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5129 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5131 ]["member-vnf-index"]
5132 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5134 ]["scaling-group-descriptor"]
5135 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5136 # for backward compatibility
5137 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5138 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5139 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5140 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5142 step
= "Getting vnfr from database"
5143 db_vnfr
= self
.db
.get_one(
5144 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5147 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5149 step
= "Getting vnfd from database"
5150 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5152 base_folder
= db_vnfd
["_admin"]["storage"]
5154 step
= "Getting scaling-group-descriptor"
5155 scaling_descriptor
= find_in_list(
5156 get_scaling_aspect(db_vnfd
),
5157 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5159 if not scaling_descriptor
:
5161 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5162 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5165 step
= "Sending scale order to VIM"
5166 # TODO check if ns is in a proper status
5168 if not db_nsr
["_admin"].get("scaling-group"):
5173 "_admin.scaling-group": [
5174 {"name": scaling_group
, "nb-scale-op": 0}
5178 admin_scale_index
= 0
5180 for admin_scale_index
, admin_scale_info
in enumerate(
5181 db_nsr
["_admin"]["scaling-group"]
5183 if admin_scale_info
["name"] == scaling_group
:
5184 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5186 else: # not found, set index one plus last element and add new entry with the name
5187 admin_scale_index
+= 1
5189 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5192 vca_scaling_info
= []
5193 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5194 if scaling_type
== "SCALE_OUT":
5195 if "aspect-delta-details" not in scaling_descriptor
:
5197 "Aspect delta details not fount in scaling descriptor {}".format(
5198 scaling_descriptor
["name"]
5201 # count if max-instance-count is reached
5202 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5204 scaling_info
["scaling_direction"] = "OUT"
5205 scaling_info
["vdu-create"] = {}
5206 scaling_info
["kdu-create"] = {}
5207 for delta
in deltas
:
5208 for vdu_delta
in delta
.get("vdu-delta", {}):
5209 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5210 # vdu_index also provides the number of instance of the targeted vdu
5211 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5212 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5216 additional_params
= (
5217 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5220 cloud_init_list
= []
5222 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5223 max_instance_count
= 10
5224 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5225 max_instance_count
= vdu_profile
.get(
5226 "max-number-of-instances", 10
5229 default_instance_num
= get_number_of_instances(
5232 instances_number
= vdu_delta
.get("number-of-instances", 1)
5233 nb_scale_op
+= instances_number
5235 new_instance_count
= nb_scale_op
+ default_instance_num
5236 # Control if new count is over max and vdu count is less than max.
5237 # Then assign new instance count
5238 if new_instance_count
> max_instance_count
> vdu_count
:
5239 instances_number
= new_instance_count
- max_instance_count
5241 instances_number
= instances_number
5243 if new_instance_count
> max_instance_count
:
5245 "reached the limit of {} (max-instance-count) "
5246 "scaling-out operations for the "
5247 "scaling-group-descriptor '{}'".format(
5248 nb_scale_op
, scaling_group
5251 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5253 # TODO Information of its own ip is not available because db_vnfr is not updated.
5254 additional_params
["OSM"] = get_osm_params(
5255 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5257 cloud_init_list
.append(
5258 self
._parse
_cloud
_init
(
5265 vca_scaling_info
.append(
5267 "osm_vdu_id": vdu_delta
["id"],
5268 "member-vnf-index": vnf_index
,
5270 "vdu_index": vdu_index
+ x
,
5273 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5274 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5275 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5276 kdu_name
= kdu_profile
["kdu-name"]
5277 resource_name
= kdu_profile
["resource-name"]
5279 # Might have different kdus in the same delta
5280 # Should have list for each kdu
5281 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5282 scaling_info
["kdu-create"][kdu_name
] = []
5284 kdur
= get_kdur(db_vnfr
, kdu_name
)
5285 if kdur
.get("helm-chart"):
5286 k8s_cluster_type
= "helm-chart-v3"
5287 self
.logger
.debug("kdur: {}".format(kdur
))
5289 kdur
.get("helm-version")
5290 and kdur
.get("helm-version") == "v2"
5292 k8s_cluster_type
= "helm-chart"
5293 raise NotImplementedError
5294 elif kdur
.get("juju-bundle"):
5295 k8s_cluster_type
= "juju-bundle"
5298 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5299 "juju-bundle. Maybe an old NBI version is running".format(
5300 db_vnfr
["member-vnf-index-ref"], kdu_name
5304 max_instance_count
= 10
5305 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5306 max_instance_count
= kdu_profile
.get(
5307 "max-number-of-instances", 10
5310 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5311 deployed_kdu
, _
= get_deployed_kdu(
5312 nsr_deployed
, kdu_name
, vnf_index
5314 if deployed_kdu
is None:
5316 "KDU '{}' for vnf '{}' not deployed".format(
5320 kdu_instance
= deployed_kdu
.get("kdu-instance")
5321 instance_num
= await self
.k8scluster_map
[
5323 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5324 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5325 "number-of-instances", 1
5328 # Control if new count is over max and instance_num is less than max.
5329 # Then assign max instance number to kdu replica count
5330 if kdu_replica_count
> max_instance_count
> instance_num
:
5331 kdu_replica_count
= max_instance_count
5332 if kdu_replica_count
> max_instance_count
:
5334 "reached the limit of {} (max-instance-count) "
5335 "scaling-out operations for the "
5336 "scaling-group-descriptor '{}'".format(
5337 instance_num
, scaling_group
5341 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5342 vca_scaling_info
.append(
5344 "osm_kdu_id": kdu_name
,
5345 "member-vnf-index": vnf_index
,
5347 "kdu_index": instance_num
+ x
- 1,
5350 scaling_info
["kdu-create"][kdu_name
].append(
5352 "member-vnf-index": vnf_index
,
5354 "k8s-cluster-type": k8s_cluster_type
,
5355 "resource-name": resource_name
,
5356 "scale": kdu_replica_count
,
5359 elif scaling_type
== "SCALE_IN":
5360 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5362 scaling_info
["scaling_direction"] = "IN"
5363 scaling_info
["vdu-delete"] = {}
5364 scaling_info
["kdu-delete"] = {}
5366 for delta
in deltas
:
5367 for vdu_delta
in delta
.get("vdu-delta", {}):
5368 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5369 min_instance_count
= 0
5370 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5371 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5372 min_instance_count
= vdu_profile
["min-number-of-instances"]
5374 default_instance_num
= get_number_of_instances(
5375 db_vnfd
, vdu_delta
["id"]
5377 instance_num
= vdu_delta
.get("number-of-instances", 1)
5378 nb_scale_op
-= instance_num
5380 new_instance_count
= nb_scale_op
+ default_instance_num
5382 if new_instance_count
< min_instance_count
< vdu_count
:
5383 instances_number
= min_instance_count
- new_instance_count
5385 instances_number
= instance_num
5387 if new_instance_count
< min_instance_count
:
5389 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5390 "scaling-group-descriptor '{}'".format(
5391 nb_scale_op
, scaling_group
5394 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5395 vca_scaling_info
.append(
5397 "osm_vdu_id": vdu_delta
["id"],
5398 "member-vnf-index": vnf_index
,
5400 "vdu_index": vdu_index
- 1 - x
,
5403 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5404 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5405 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5406 kdu_name
= kdu_profile
["kdu-name"]
5407 resource_name
= kdu_profile
["resource-name"]
5409 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5410 scaling_info
["kdu-delete"][kdu_name
] = []
5412 kdur
= get_kdur(db_vnfr
, kdu_name
)
5413 if kdur
.get("helm-chart"):
5414 k8s_cluster_type
= "helm-chart-v3"
5415 self
.logger
.debug("kdur: {}".format(kdur
))
5417 kdur
.get("helm-version")
5418 and kdur
.get("helm-version") == "v2"
5420 k8s_cluster_type
= "helm-chart"
5421 raise NotImplementedError
5422 elif kdur
.get("juju-bundle"):
5423 k8s_cluster_type
= "juju-bundle"
5426 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5427 "juju-bundle. Maybe an old NBI version is running".format(
5428 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5432 min_instance_count
= 0
5433 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5434 min_instance_count
= kdu_profile
["min-number-of-instances"]
5436 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5437 deployed_kdu
, _
= get_deployed_kdu(
5438 nsr_deployed
, kdu_name
, vnf_index
5440 if deployed_kdu
is None:
5442 "KDU '{}' for vnf '{}' not deployed".format(
5446 kdu_instance
= deployed_kdu
.get("kdu-instance")
5447 instance_num
= await self
.k8scluster_map
[
5449 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5450 kdu_replica_count
= instance_num
- kdu_delta
.get(
5451 "number-of-instances", 1
5454 if kdu_replica_count
< min_instance_count
< instance_num
:
5455 kdu_replica_count
= min_instance_count
5456 if kdu_replica_count
< min_instance_count
:
5458 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5459 "scaling-group-descriptor '{}'".format(
5460 instance_num
, scaling_group
5464 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5465 vca_scaling_info
.append(
5467 "osm_kdu_id": kdu_name
,
5468 "member-vnf-index": vnf_index
,
5470 "kdu_index": instance_num
- x
- 1,
5473 scaling_info
["kdu-delete"][kdu_name
].append(
5475 "member-vnf-index": vnf_index
,
5477 "k8s-cluster-type": k8s_cluster_type
,
5478 "resource-name": resource_name
,
5479 "scale": kdu_replica_count
,
5483 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5484 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5485 if scaling_info
["scaling_direction"] == "IN":
5486 for vdur
in reversed(db_vnfr
["vdur"]):
5487 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5488 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5489 scaling_info
["vdu"].append(
5491 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5492 "vdu_id": vdur
["vdu-id-ref"],
5496 for interface
in vdur
["interfaces"]:
5497 scaling_info
["vdu"][-1]["interface"].append(
5499 "name": interface
["name"],
5500 "ip_address": interface
["ip-address"],
5501 "mac_address": interface
.get("mac-address"),
5504 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5507 step
= "Executing pre-scale vnf-config-primitive"
5508 if scaling_descriptor
.get("scaling-config-action"):
5509 for scaling_config_action
in scaling_descriptor
[
5510 "scaling-config-action"
5513 scaling_config_action
.get("trigger") == "pre-scale-in"
5514 and scaling_type
== "SCALE_IN"
5516 scaling_config_action
.get("trigger") == "pre-scale-out"
5517 and scaling_type
== "SCALE_OUT"
5519 vnf_config_primitive
= scaling_config_action
[
5520 "vnf-config-primitive-name-ref"
5522 step
= db_nslcmop_update
[
5524 ] = "executing pre-scale scaling-config-action '{}'".format(
5525 vnf_config_primitive
5528 # look for primitive
5529 for config_primitive
in (
5530 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5531 ).get("config-primitive", ()):
5532 if config_primitive
["name"] == vnf_config_primitive
:
5536 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5537 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5538 "primitive".format(scaling_group
, vnf_config_primitive
)
5541 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5542 if db_vnfr
.get("additionalParamsForVnf"):
5543 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5545 scale_process
= "VCA"
5546 db_nsr_update
["config-status"] = "configuring pre-scaling"
5547 primitive_params
= self
._map
_primitive
_params
(
5548 config_primitive
, {}, vnfr_params
5551 # Pre-scale retry check: Check if this sub-operation has been executed before
5552 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5555 vnf_config_primitive
,
5559 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5560 # Skip sub-operation
5561 result
= "COMPLETED"
5562 result_detail
= "Done"
5565 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5566 vnf_config_primitive
, result
, result_detail
5570 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5571 # New sub-operation: Get index of this sub-operation
5573 len(db_nslcmop
.get("_admin", {}).get("operations"))
5578 + "vnf_config_primitive={} New sub-operation".format(
5579 vnf_config_primitive
5583 # retry: Get registered params for this existing sub-operation
5584 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5587 vnf_index
= op
.get("member_vnf_index")
5588 vnf_config_primitive
= op
.get("primitive")
5589 primitive_params
= op
.get("primitive_params")
5592 + "vnf_config_primitive={} Sub-operation retry".format(
5593 vnf_config_primitive
5596 # Execute the primitive, either with new (first-time) or registered (reintent) args
5597 ee_descriptor_id
= config_primitive
.get(
5598 "execution-environment-ref"
5600 primitive_name
= config_primitive
.get(
5601 "execution-environment-primitive", vnf_config_primitive
5603 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5604 nsr_deployed
["VCA"],
5605 member_vnf_index
=vnf_index
,
5607 vdu_count_index
=None,
5608 ee_descriptor_id
=ee_descriptor_id
,
5610 result
, result_detail
= await self
._ns
_execute
_primitive
(
5619 + "vnf_config_primitive={} Done with result {} {}".format(
5620 vnf_config_primitive
, result
, result_detail
5623 # Update operationState = COMPLETED | FAILED
5624 self
._update
_suboperation
_status
(
5625 db_nslcmop
, op_index
, result
, result_detail
5628 if result
== "FAILED":
5629 raise LcmException(result_detail
)
5630 db_nsr_update
["config-status"] = old_config_status
5631 scale_process
= None
5635 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5638 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5641 # SCALE-IN VCA - BEGIN
5642 if vca_scaling_info
:
5643 step
= db_nslcmop_update
[
5645 ] = "Deleting the execution environments"
5646 scale_process
= "VCA"
5647 for vca_info
in vca_scaling_info
:
5648 if vca_info
["type"] == "delete":
5649 member_vnf_index
= str(vca_info
["member-vnf-index"])
5651 logging_text
+ "vdu info: {}".format(vca_info
)
5653 if vca_info
.get("osm_vdu_id"):
5654 vdu_id
= vca_info
["osm_vdu_id"]
5655 vdu_index
= int(vca_info
["vdu_index"])
5658 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5659 member_vnf_index
, vdu_id
, vdu_index
5663 kdu_id
= vca_info
["osm_kdu_id"]
5666 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5667 member_vnf_index
, kdu_id
, vdu_index
5669 stage
[2] = step
= "Scaling in VCA"
5670 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5671 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5672 config_update
= db_nsr
["configurationStatus"]
5673 for vca_index
, vca
in enumerate(vca_update
):
5675 (vca
or vca
.get("ee_id"))
5676 and vca
["member-vnf-index"] == member_vnf_index
5677 and vca
["vdu_count_index"] == vdu_index
5679 if vca
.get("vdu_id"):
5680 config_descriptor
= get_configuration(
5681 db_vnfd
, vca
.get("vdu_id")
5683 elif vca
.get("kdu_name"):
5684 config_descriptor
= get_configuration(
5685 db_vnfd
, vca
.get("kdu_name")
5688 config_descriptor
= get_configuration(
5689 db_vnfd
, db_vnfd
["id"]
5691 operation_params
= (
5692 db_nslcmop
.get("operationParams") or {}
5694 exec_terminate_primitives
= not operation_params
.get(
5695 "skip_terminate_primitives"
5696 ) and vca
.get("needed_terminate")
5697 task
= asyncio
.ensure_future(
5706 exec_primitives
=exec_terminate_primitives
,
5710 timeout
=self
.timeout_charm_delete
,
5713 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5716 del vca_update
[vca_index
]
5717 del config_update
[vca_index
]
5718 # wait for pending tasks of terminate primitives
5722 + "Waiting for tasks {}".format(
5723 list(tasks_dict_info
.keys())
5726 error_list
= await self
._wait
_for
_tasks
(
5730 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5735 tasks_dict_info
.clear()
5737 raise LcmException("; ".join(error_list
))
5739 db_vca_and_config_update
= {
5740 "_admin.deployed.VCA": vca_update
,
5741 "configurationStatus": config_update
,
5744 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5746 scale_process
= None
5747 # SCALE-IN VCA - END
5750 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5751 scale_process
= "RO"
5752 if self
.ro_config
.get("ng"):
5753 await self
._scale
_ng
_ro
(
5754 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5756 scaling_info
.pop("vdu-create", None)
5757 scaling_info
.pop("vdu-delete", None)
5759 scale_process
= None
5763 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5764 scale_process
= "KDU"
5765 await self
._scale
_kdu
(
5766 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5768 scaling_info
.pop("kdu-create", None)
5769 scaling_info
.pop("kdu-delete", None)
5771 scale_process
= None
5775 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5777 # SCALE-UP VCA - BEGIN
5778 if vca_scaling_info
:
5779 step
= db_nslcmop_update
[
5781 ] = "Creating new execution environments"
5782 scale_process
= "VCA"
5783 for vca_info
in vca_scaling_info
:
5784 if vca_info
["type"] == "create":
5785 member_vnf_index
= str(vca_info
["member-vnf-index"])
5787 logging_text
+ "vdu info: {}".format(vca_info
)
5789 vnfd_id
= db_vnfr
["vnfd-ref"]
5790 if vca_info
.get("osm_vdu_id"):
5791 vdu_index
= int(vca_info
["vdu_index"])
5792 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5793 if db_vnfr
.get("additionalParamsForVnf"):
5794 deploy_params
.update(
5796 db_vnfr
["additionalParamsForVnf"].copy()
5799 descriptor_config
= get_configuration(
5800 db_vnfd
, db_vnfd
["id"]
5802 if descriptor_config
:
5807 logging_text
=logging_text
5808 + "member_vnf_index={} ".format(member_vnf_index
),
5811 nslcmop_id
=nslcmop_id
,
5817 member_vnf_index
=member_vnf_index
,
5818 vdu_index
=vdu_index
,
5820 deploy_params
=deploy_params
,
5821 descriptor_config
=descriptor_config
,
5822 base_folder
=base_folder
,
5823 task_instantiation_info
=tasks_dict_info
,
5826 vdu_id
= vca_info
["osm_vdu_id"]
5827 vdur
= find_in_list(
5828 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5830 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5831 if vdur
.get("additionalParams"):
5832 deploy_params_vdu
= parse_yaml_strings(
5833 vdur
["additionalParams"]
5836 deploy_params_vdu
= deploy_params
5837 deploy_params_vdu
["OSM"] = get_osm_params(
5838 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5840 if descriptor_config
:
5845 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5846 member_vnf_index
, vdu_id
, vdu_index
5848 stage
[2] = step
= "Scaling out VCA"
5849 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5851 logging_text
=logging_text
5852 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5853 member_vnf_index
, vdu_id
, vdu_index
5857 nslcmop_id
=nslcmop_id
,
5863 member_vnf_index
=member_vnf_index
,
5864 vdu_index
=vdu_index
,
5866 deploy_params
=deploy_params_vdu
,
5867 descriptor_config
=descriptor_config
,
5868 base_folder
=base_folder
,
5869 task_instantiation_info
=tasks_dict_info
,
5873 kdu_name
= vca_info
["osm_kdu_id"]
5874 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5875 if descriptor_config
:
5877 kdu_index
= int(vca_info
["kdu_index"])
5881 for x
in db_vnfr
["kdur"]
5882 if x
["kdu-name"] == kdu_name
5884 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5885 if kdur
.get("additionalParams"):
5886 deploy_params_kdu
= parse_yaml_strings(
5887 kdur
["additionalParams"]
5891 logging_text
=logging_text
,
5894 nslcmop_id
=nslcmop_id
,
5900 member_vnf_index
=member_vnf_index
,
5901 vdu_index
=kdu_index
,
5903 deploy_params
=deploy_params_kdu
,
5904 descriptor_config
=descriptor_config
,
5905 base_folder
=base_folder
,
5906 task_instantiation_info
=tasks_dict_info
,
5909 # SCALE-UP VCA - END
5910 scale_process
= None
5913 # execute primitive service POST-SCALING
5914 step
= "Executing post-scale vnf-config-primitive"
5915 if scaling_descriptor
.get("scaling-config-action"):
5916 for scaling_config_action
in scaling_descriptor
[
5917 "scaling-config-action"
5920 scaling_config_action
.get("trigger") == "post-scale-in"
5921 and scaling_type
== "SCALE_IN"
5923 scaling_config_action
.get("trigger") == "post-scale-out"
5924 and scaling_type
== "SCALE_OUT"
5926 vnf_config_primitive
= scaling_config_action
[
5927 "vnf-config-primitive-name-ref"
5929 step
= db_nslcmop_update
[
5931 ] = "executing post-scale scaling-config-action '{}'".format(
5932 vnf_config_primitive
5935 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5936 if db_vnfr
.get("additionalParamsForVnf"):
5937 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5939 # look for primitive
5940 for config_primitive
in (
5941 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5942 ).get("config-primitive", ()):
5943 if config_primitive
["name"] == vnf_config_primitive
:
5947 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
5948 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
5949 "config-primitive".format(
5950 scaling_group
, vnf_config_primitive
5953 scale_process
= "VCA"
5954 db_nsr_update
["config-status"] = "configuring post-scaling"
5955 primitive_params
= self
._map
_primitive
_params
(
5956 config_primitive
, {}, vnfr_params
5959 # Post-scale retry check: Check if this sub-operation has been executed before
5960 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5963 vnf_config_primitive
,
5967 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5968 # Skip sub-operation
5969 result
= "COMPLETED"
5970 result_detail
= "Done"
5973 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5974 vnf_config_primitive
, result
, result_detail
5978 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5979 # New sub-operation: Get index of this sub-operation
5981 len(db_nslcmop
.get("_admin", {}).get("operations"))
5986 + "vnf_config_primitive={} New sub-operation".format(
5987 vnf_config_primitive
5991 # retry: Get registered params for this existing sub-operation
5992 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5995 vnf_index
= op
.get("member_vnf_index")
5996 vnf_config_primitive
= op
.get("primitive")
5997 primitive_params
= op
.get("primitive_params")
6000 + "vnf_config_primitive={} Sub-operation retry".format(
6001 vnf_config_primitive
6004 # Execute the primitive, either with new (first-time) or registered (reintent) args
6005 ee_descriptor_id
= config_primitive
.get(
6006 "execution-environment-ref"
6008 primitive_name
= config_primitive
.get(
6009 "execution-environment-primitive", vnf_config_primitive
6011 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6012 nsr_deployed
["VCA"],
6013 member_vnf_index
=vnf_index
,
6015 vdu_count_index
=None,
6016 ee_descriptor_id
=ee_descriptor_id
,
6018 result
, result_detail
= await self
._ns
_execute
_primitive
(
6027 + "vnf_config_primitive={} Done with result {} {}".format(
6028 vnf_config_primitive
, result
, result_detail
6031 # Update operationState = COMPLETED | FAILED
6032 self
._update
_suboperation
_status
(
6033 db_nslcmop
, op_index
, result
, result_detail
6036 if result
== "FAILED":
6037 raise LcmException(result_detail
)
6038 db_nsr_update
["config-status"] = old_config_status
6039 scale_process
= None
6044 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6045 db_nsr_update
["operational-status"] = (
6047 if old_operational_status
== "failed"
6048 else old_operational_status
6050 db_nsr_update
["config-status"] = old_config_status
6053 ROclient
.ROClientException
,
6058 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6060 except asyncio
.CancelledError
:
6062 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6064 exc
= "Operation was cancelled"
6065 except Exception as e
:
6066 exc
= traceback
.format_exc()
6067 self
.logger
.critical(
6068 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6072 self
._write
_ns
_status
(
6075 current_operation
="IDLE",
6076 current_operation_id
=None,
6079 stage
[1] = "Waiting for instantiate pending tasks."
6080 self
.logger
.debug(logging_text
+ stage
[1])
6081 exc
= await self
._wait
_for
_tasks
(
6084 self
.timeout_ns_deploy
,
6092 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6093 nslcmop_operation_state
= "FAILED"
6095 db_nsr_update
["operational-status"] = old_operational_status
6096 db_nsr_update
["config-status"] = old_config_status
6097 db_nsr_update
["detailed-status"] = ""
6099 if "VCA" in scale_process
:
6100 db_nsr_update
["config-status"] = "failed"
6101 if "RO" in scale_process
:
6102 db_nsr_update
["operational-status"] = "failed"
6105 ] = "FAILED scaling nslcmop={} {}: {}".format(
6106 nslcmop_id
, step
, exc
6109 error_description_nslcmop
= None
6110 nslcmop_operation_state
= "COMPLETED"
6111 db_nslcmop_update
["detailed-status"] = "Done"
6113 self
._write
_op
_status
(
6116 error_message
=error_description_nslcmop
,
6117 operation_state
=nslcmop_operation_state
,
6118 other_update
=db_nslcmop_update
,
6121 self
._write
_ns
_status
(
6124 current_operation
="IDLE",
6125 current_operation_id
=None,
6126 other_update
=db_nsr_update
,
6129 if nslcmop_operation_state
:
6133 "nslcmop_id": nslcmop_id
,
6134 "operationState": nslcmop_operation_state
,
6136 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6137 except Exception as e
:
6139 logging_text
+ "kafka_write notification Exception {}".format(e
)
6141 self
.logger
.debug(logging_text
+ "Exit")
6142 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6144 async def _scale_kdu(
6145 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6147 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6148 for kdu_name
in _scaling_info
:
6149 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6150 deployed_kdu
, index
= get_deployed_kdu(
6151 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6153 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6154 kdu_instance
= deployed_kdu
["kdu-instance"]
6155 scale
= int(kdu_scaling_info
["scale"])
6156 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6159 "collection": "nsrs",
6160 "filter": {"_id": nsr_id
},
6161 "path": "_admin.deployed.K8s.{}".format(index
),
6164 step
= "scaling application {}".format(
6165 kdu_scaling_info
["resource-name"]
6167 self
.logger
.debug(logging_text
+ step
)
6169 if kdu_scaling_info
["type"] == "delete":
6170 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6173 and kdu_config
.get("terminate-config-primitive")
6174 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6176 terminate_config_primitive_list
= kdu_config
.get(
6177 "terminate-config-primitive"
6179 terminate_config_primitive_list
.sort(
6180 key
=lambda val
: int(val
["seq"])
6184 terminate_config_primitive
6185 ) in terminate_config_primitive_list
:
6186 primitive_params_
= self
._map
_primitive
_params
(
6187 terminate_config_primitive
, {}, {}
6189 step
= "execute terminate config primitive"
6190 self
.logger
.debug(logging_text
+ step
)
6191 await asyncio
.wait_for(
6192 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6193 cluster_uuid
=cluster_uuid
,
6194 kdu_instance
=kdu_instance
,
6195 primitive_name
=terminate_config_primitive
["name"],
6196 params
=primitive_params_
,
6203 await asyncio
.wait_for(
6204 self
.k8scluster_map
[k8s_cluster_type
].scale(
6207 kdu_scaling_info
["resource-name"],
6210 timeout
=self
.timeout_vca_on_error
,
6213 if kdu_scaling_info
["type"] == "create":
6214 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6217 and kdu_config
.get("initial-config-primitive")
6218 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6220 initial_config_primitive_list
= kdu_config
.get(
6221 "initial-config-primitive"
6223 initial_config_primitive_list
.sort(
6224 key
=lambda val
: int(val
["seq"])
6227 for initial_config_primitive
in initial_config_primitive_list
:
6228 primitive_params_
= self
._map
_primitive
_params
(
6229 initial_config_primitive
, {}, {}
6231 step
= "execute initial config primitive"
6232 self
.logger
.debug(logging_text
+ step
)
6233 await asyncio
.wait_for(
6234 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6235 cluster_uuid
=cluster_uuid
,
6236 kdu_instance
=kdu_instance
,
6237 primitive_name
=initial_config_primitive
["name"],
6238 params
=primitive_params_
,
6245 async def _scale_ng_ro(
6246 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6248 nsr_id
= db_nslcmop
["nsInstanceId"]
6249 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6252 # read from db: vnfd's for every vnf
6255 # for each vnf in ns, read vnfd
6256 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6257 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6258 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6259 # if we haven't this vnfd, read it from db
6260 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6262 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6263 db_vnfds
.append(vnfd
)
6264 n2vc_key
= self
.n2vc
.get_public_key()
6265 n2vc_key_list
= [n2vc_key
]
6268 vdu_scaling_info
.get("vdu-create"),
6269 vdu_scaling_info
.get("vdu-delete"),
6272 # db_vnfr has been updated, update db_vnfrs to use it
6273 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6274 await self
._instantiate
_ng
_ro
(
6284 start_deploy
=time(),
6285 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6287 if vdu_scaling_info
.get("vdu-delete"):
6289 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6292 async def add_prometheus_metrics(
6293 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6295 if not self
.prometheus
:
6297 # look if exist a file called 'prometheus*.j2' and
6298 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6302 for f
in artifact_content
6303 if f
.startswith("prometheus") and f
.endswith(".j2")
6309 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6313 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6314 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6316 vnfr_id
= vnfr_id
.replace("-", "")
6318 "JOB_NAME": vnfr_id
,
6319 "TARGET_IP": target_ip
,
6320 "EXPORTER_POD_IP": host_name
,
6321 "EXPORTER_POD_PORT": host_port
,
6323 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
6324 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6325 for job
in job_list
:
6327 not isinstance(job
.get("job_name"), str)
6328 or vnfr_id
not in job
["job_name"]
6330 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6331 job
["nsr_id"] = nsr_id
6332 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
6333 if await self
.prometheus
.update(job_dict
):
6334 return list(job_dict
.keys())
6336 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6338 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6340 :param: vim_account_id: VIM Account ID
6342 :return: (cloud_name, cloud_credential)
6344 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6345 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6347 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6349 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6351 :param: vim_account_id: VIM Account ID
6353 :return: (cloud_name, cloud_credential)
6355 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6356 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")