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
22 import logging
.handlers
33 from osm_lcm
import ROclient
34 from osm_lcm
.data_utils
.nsr
import get_deployed_kdu
35 from osm_lcm
.ng_ro
import NgRoClient
, NgRoException
36 from osm_lcm
.lcm_utils
import (
44 from osm_lcm
.data_utils
.nsd
import get_vnf_profiles
45 from osm_lcm
.data_utils
.vnfd
import (
48 get_ee_sorted_initial_config_primitive_list
,
49 get_ee_sorted_terminate_config_primitive_list
,
51 get_virtual_link_profiles
,
56 get_number_of_instances
,
60 from osm_lcm
.data_utils
.list_utils
import find_in_list
61 from osm_lcm
.data_utils
.vnfr
import get_osm_params
, get_vdur_index
, get_kdur
62 from osm_lcm
.data_utils
.dict_utils
import parse_yaml_strings
63 from osm_lcm
.data_utils
.database
.vim_account
import VimAccountDB
64 from n2vc
.k8s_helm_conn
import K8sHelmConnector
65 from n2vc
.k8s_helm3_conn
import K8sHelm3Connector
66 from n2vc
.k8s_juju_conn
import K8sJujuConnector
68 from osm_common
.dbbase
import DbException
69 from osm_common
.fsbase
import FsException
71 from osm_lcm
.data_utils
.database
.database
import Database
72 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
74 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
75 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
77 from osm_lcm
.lcm_helm_conn
import LCMHelmConn
79 from copy
import copy
, deepcopy
81 from uuid
import uuid4
83 from random
import randint
85 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
89 timeout_vca_on_error
= (
91 ) # Time for charm from first time at blocked,error status to mark as failed
92 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
93 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
94 timeout_charm_delete
= 10 * 60
95 timeout_primitive
= 30 * 60 # timeout for primitive execution
96 timeout_progress_primitive
= (
98 ) # timeout for some progress in a primitive execution
100 SUBOPERATION_STATUS_NOT_FOUND
= -1
101 SUBOPERATION_STATUS_NEW
= -2
102 SUBOPERATION_STATUS_SKIP
= -3
103 task_name_deploy_vca
= "Deploying VCA"
105 def __init__(self
, msg
, lcm_tasks
, config
, loop
, prometheus
=None):
107 Init, Connect to database, filesystem storage, and messaging
108 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
111 super().__init
__(msg
=msg
, logger
=logging
.getLogger("lcm.ns"))
113 self
.db
= Database().instance
.db
114 self
.fs
= Filesystem().instance
.fs
116 self
.lcm_tasks
= lcm_tasks
117 self
.timeout
= config
["timeout"]
118 self
.ro_config
= config
["ro_config"]
119 self
.ng_ro
= config
["ro_config"].get("ng")
120 self
.vca_config
= config
["VCA"].copy()
122 # create N2VC connector
123 self
.n2vc
= N2VCJujuConnector(
126 on_update_db
=self
._on
_update
_n
2vc
_db
,
131 self
.conn_helm_ee
= LCMHelmConn(
134 vca_config
=self
.vca_config
,
135 on_update_db
=self
._on
_update
_n
2vc
_db
,
138 self
.k8sclusterhelm2
= K8sHelmConnector(
139 kubectl_command
=self
.vca_config
.get("kubectlpath"),
140 helm_command
=self
.vca_config
.get("helmpath"),
147 self
.k8sclusterhelm3
= K8sHelm3Connector(
148 kubectl_command
=self
.vca_config
.get("kubectlpath"),
149 helm_command
=self
.vca_config
.get("helm3path"),
156 self
.k8sclusterjuju
= K8sJujuConnector(
157 kubectl_command
=self
.vca_config
.get("kubectlpath"),
158 juju_command
=self
.vca_config
.get("jujupath"),
161 on_update_db
=self
._on
_update
_k
8s
_db
,
166 self
.k8scluster_map
= {
167 "helm-chart": self
.k8sclusterhelm2
,
168 "helm-chart-v3": self
.k8sclusterhelm3
,
169 "chart": self
.k8sclusterhelm3
,
170 "juju-bundle": self
.k8sclusterjuju
,
171 "juju": self
.k8sclusterjuju
,
175 "lxc_proxy_charm": self
.n2vc
,
176 "native_charm": self
.n2vc
,
177 "k8s_proxy_charm": self
.n2vc
,
178 "helm": self
.conn_helm_ee
,
179 "helm-v3": self
.conn_helm_ee
,
182 self
.prometheus
= prometheus
185 self
.RO
= NgRoClient(self
.loop
, **self
.ro_config
)
188 def increment_ip_mac(ip_mac
, vm_index
=1):
189 if not isinstance(ip_mac
, str):
192 # try with ipv4 look for last dot
193 i
= ip_mac
.rfind(".")
196 return "{}{}".format(ip_mac
[:i
], int(ip_mac
[i
:]) + vm_index
)
197 # try with ipv6 or mac look for last colon. Operate in hex
198 i
= ip_mac
.rfind(":")
201 # format in hex, len can be 2 for mac or 4 for ipv6
202 return ("{}{:0" + str(len(ip_mac
) - i
) + "x}").format(
203 ip_mac
[:i
], int(ip_mac
[i
:], 16) + vm_index
209 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
211 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
214 # TODO filter RO descriptor fields...
218 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
219 db_dict
["deploymentStatus"] = ro_descriptor
220 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
222 except Exception as e
:
224 "Cannot write database RO deployment for ns={} -> {}".format(nsrs_id
, e
)
227 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
, vca_id
=None):
229 # remove last dot from path (if exists)
230 if path
.endswith("."):
233 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
234 # .format(table, filter, path, updated_data))
237 nsr_id
= filter.get("_id")
239 # read ns record from database
240 nsr
= self
.db
.get_one(table
="nsrs", q_filter
=filter)
241 current_ns_status
= nsr
.get("nsState")
243 # get vca status for NS
244 status_dict
= await self
.n2vc
.get_status(
245 namespace
="." + nsr_id
, yaml_format
=False, vca_id
=vca_id
250 db_dict
["vcaStatus"] = status_dict
251 await self
.n2vc
.update_vca_status(db_dict
["vcaStatus"], vca_id
=vca_id
)
253 # update configurationStatus for this VCA
255 vca_index
= int(path
[path
.rfind(".") + 1 :])
258 target_dict
=nsr
, key_list
=("_admin", "deployed", "VCA")
260 vca_status
= vca_list
[vca_index
].get("status")
262 configuration_status_list
= nsr
.get("configurationStatus")
263 config_status
= configuration_status_list
[vca_index
].get("status")
265 if config_status
== "BROKEN" and vca_status
!= "failed":
266 db_dict
["configurationStatus"][vca_index
] = "READY"
267 elif config_status
!= "BROKEN" and vca_status
== "failed":
268 db_dict
["configurationStatus"][vca_index
] = "BROKEN"
269 except Exception as e
:
270 # not update configurationStatus
271 self
.logger
.debug("Error updating vca_index (ignore): {}".format(e
))
273 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
274 # if nsState = 'DEGRADED' check if all is OK
276 if current_ns_status
in ("READY", "DEGRADED"):
277 error_description
= ""
279 if status_dict
.get("machines"):
280 for machine_id
in status_dict
.get("machines"):
281 machine
= status_dict
.get("machines").get(machine_id
)
282 # check machine agent-status
283 if machine
.get("agent-status"):
284 s
= machine
.get("agent-status").get("status")
287 error_description
+= (
288 "machine {} agent-status={} ; ".format(
292 # check machine instance status
293 if machine
.get("instance-status"):
294 s
= machine
.get("instance-status").get("status")
297 error_description
+= (
298 "machine {} instance-status={} ; ".format(
303 if status_dict
.get("applications"):
304 for app_id
in status_dict
.get("applications"):
305 app
= status_dict
.get("applications").get(app_id
)
306 # check application status
307 if app
.get("status"):
308 s
= app
.get("status").get("status")
311 error_description
+= (
312 "application {} status={} ; ".format(app_id
, s
)
315 if error_description
:
316 db_dict
["errorDescription"] = error_description
317 if current_ns_status
== "READY" and is_degraded
:
318 db_dict
["nsState"] = "DEGRADED"
319 if current_ns_status
== "DEGRADED" and not is_degraded
:
320 db_dict
["nsState"] = "READY"
323 self
.update_db_2("nsrs", nsr_id
, db_dict
)
325 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
327 except Exception as e
:
328 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
330 async def _on_update_k8s_db(
331 self
, cluster_uuid
, kdu_instance
, filter=None, vca_id
=None
334 Updating vca status in NSR record
335 :param cluster_uuid: UUID of a k8s cluster
336 :param kdu_instance: The unique name of the KDU instance
337 :param filter: To get nsr_id
341 # self.logger.debug("_on_update_k8s_db(cluster_uuid={}, kdu_instance={}, filter={}"
342 # .format(cluster_uuid, kdu_instance, filter))
345 nsr_id
= filter.get("_id")
347 # get vca status for NS
348 vca_status
= await self
.k8sclusterjuju
.status_kdu(
351 complete_status
=True,
357 db_dict
["vcaStatus"] = {nsr_id
: vca_status
}
359 await self
.k8sclusterjuju
.update_vca_status(
360 db_dict
["vcaStatus"],
366 self
.update_db_2("nsrs", nsr_id
, db_dict
)
368 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
370 except Exception as e
:
371 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
374 def _parse_cloud_init(cloud_init_text
, additional_params
, vnfd_id
, vdu_id
):
376 env
= Environment(undefined
=StrictUndefined
)
377 template
= env
.from_string(cloud_init_text
)
378 return template
.render(additional_params
or {})
379 except UndefinedError
as e
:
381 "Variable {} at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
382 "file, must be provided in the instantiation parameters inside the "
383 "'additionalParamsForVnf/Vdu' block".format(e
, vnfd_id
, vdu_id
)
385 except (TemplateError
, TemplateNotFound
) as e
:
387 "Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".format(
392 def _get_vdu_cloud_init_content(self
, vdu
, vnfd
):
393 cloud_init_content
= cloud_init_file
= None
395 if vdu
.get("cloud-init-file"):
396 base_folder
= vnfd
["_admin"]["storage"]
397 cloud_init_file
= "{}/{}/cloud_init/{}".format(
398 base_folder
["folder"],
399 base_folder
["pkg-dir"],
400 vdu
["cloud-init-file"],
402 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
403 cloud_init_content
= ci_file
.read()
404 elif vdu
.get("cloud-init"):
405 cloud_init_content
= vdu
["cloud-init"]
407 return cloud_init_content
408 except FsException
as e
:
410 "Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".format(
411 vnfd
["id"], vdu
["id"], cloud_init_file
, e
415 def _get_vdu_additional_params(self
, db_vnfr
, vdu_id
):
417 vdur
for vdur
in db_vnfr
.get("vdur") if vdu_id
== vdur
["vdu-id-ref"]
419 additional_params
= vdur
.get("additionalParams")
420 return parse_yaml_strings(additional_params
)
422 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
424 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
425 :param vnfd: input vnfd
426 :param new_id: overrides vnf id if provided
427 :param additionalParams: Instantiation params for VNFs provided
428 :param nsrId: Id of the NSR
429 :return: copy of vnfd
431 vnfd_RO
= deepcopy(vnfd
)
432 # remove unused by RO configuration, monitoring, scaling and internal keys
433 vnfd_RO
.pop("_id", None)
434 vnfd_RO
.pop("_admin", None)
435 vnfd_RO
.pop("monitoring-param", None)
436 vnfd_RO
.pop("scaling-group-descriptor", None)
437 vnfd_RO
.pop("kdu", None)
438 vnfd_RO
.pop("k8s-cluster", None)
440 vnfd_RO
["id"] = new_id
442 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
443 for vdu
in get_iterable(vnfd_RO
, "vdu"):
444 vdu
.pop("cloud-init-file", None)
445 vdu
.pop("cloud-init", None)
449 def ip_profile_2_RO(ip_profile
):
450 RO_ip_profile
= deepcopy(ip_profile
)
451 if "dns-server" in RO_ip_profile
:
452 if isinstance(RO_ip_profile
["dns-server"], list):
453 RO_ip_profile
["dns-address"] = []
454 for ds
in RO_ip_profile
.pop("dns-server"):
455 RO_ip_profile
["dns-address"].append(ds
["address"])
457 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
458 if RO_ip_profile
.get("ip-version") == "ipv4":
459 RO_ip_profile
["ip-version"] = "IPv4"
460 if RO_ip_profile
.get("ip-version") == "ipv6":
461 RO_ip_profile
["ip-version"] = "IPv6"
462 if "dhcp-params" in RO_ip_profile
:
463 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
466 def _get_ro_vim_id_for_vim_account(self
, vim_account
):
467 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
468 if db_vim
["_admin"]["operationalState"] != "ENABLED":
470 "VIM={} is not available. operationalState={}".format(
471 vim_account
, db_vim
["_admin"]["operationalState"]
474 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
477 def get_ro_wim_id_for_wim_account(self
, wim_account
):
478 if isinstance(wim_account
, str):
479 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
480 if db_wim
["_admin"]["operationalState"] != "ENABLED":
482 "WIM={} is not available. operationalState={}".format(
483 wim_account
, db_wim
["_admin"]["operationalState"]
486 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
491 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None, mark_delete
=False):
493 db_vdu_push_list
= []
494 db_update
= {"_admin.modified": time()}
496 for vdu_id
, vdu_count
in vdu_create
.items():
500 for vdur
in reversed(db_vnfr
["vdur"])
501 if vdur
["vdu-id-ref"] == vdu_id
507 "Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
512 for count
in range(vdu_count
):
513 vdur_copy
= deepcopy(vdur
)
514 vdur_copy
["status"] = "BUILD"
515 vdur_copy
["status-detailed"] = None
516 vdur_copy
["ip-address"] = None
517 vdur_copy
["_id"] = str(uuid4())
518 vdur_copy
["count-index"] += count
+ 1
519 vdur_copy
["id"] = "{}-{}".format(
520 vdur_copy
["vdu-id-ref"], vdur_copy
["count-index"]
522 vdur_copy
.pop("vim_info", None)
523 for iface
in vdur_copy
["interfaces"]:
524 if iface
.get("fixed-ip"):
525 iface
["ip-address"] = self
.increment_ip_mac(
526 iface
["ip-address"], count
+ 1
529 iface
.pop("ip-address", None)
530 if iface
.get("fixed-mac"):
531 iface
["mac-address"] = self
.increment_ip_mac(
532 iface
["mac-address"], count
+ 1
535 iface
.pop("mac-address", None)
538 ) # only first vdu can be managment of vnf
539 db_vdu_push_list
.append(vdur_copy
)
540 # self.logger.debug("scale out, adding vdu={}".format(vdur_copy))
542 for vdu_id
, vdu_count
in vdu_delete
.items():
544 indexes_to_delete
= [
546 for iv
in enumerate(db_vnfr
["vdur"])
547 if iv
[1]["vdu-id-ref"] == vdu_id
551 "vdur.{}.status".format(i
): "DELETING"
552 for i
in indexes_to_delete
[-vdu_count
:]
556 # it must be deleted one by one because common.db does not allow otherwise
559 for v
in reversed(db_vnfr
["vdur"])
560 if v
["vdu-id-ref"] == vdu_id
562 for vdu
in vdus_to_delete
[:vdu_count
]:
565 {"_id": db_vnfr
["_id"]},
567 pull
={"vdur": {"_id": vdu
["_id"]}},
569 db_push
= {"vdur": db_vdu_push_list
} if db_vdu_push_list
else None
570 self
.db
.set_one("vnfrs", {"_id": db_vnfr
["_id"]}, db_update
, push_list
=db_push
)
571 # modify passed dictionary db_vnfr
572 db_vnfr_
= self
.db
.get_one("vnfrs", {"_id": db_vnfr
["_id"]})
573 db_vnfr
["vdur"] = db_vnfr_
["vdur"]
575 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
577 Updates database nsr with the RO info for the created vld
578 :param ns_update_nsr: dictionary to be filled with the updated info
579 :param db_nsr: content of db_nsr. This is also modified
580 :param nsr_desc_RO: nsr descriptor from RO
581 :return: Nothing, LcmException is raised on errors
584 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
585 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
586 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
588 vld
["vim-id"] = net_RO
.get("vim_net_id")
589 vld
["name"] = net_RO
.get("vim_name")
590 vld
["status"] = net_RO
.get("status")
591 vld
["status-detailed"] = net_RO
.get("error_msg")
592 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
596 "ns_update_nsr: Not found vld={} at RO info".format(vld
["id"])
599 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
601 for db_vnfr
in db_vnfrs
.values():
602 vnfr_update
= {"status": "ERROR"}
603 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
604 if "status" not in vdur
:
605 vdur
["status"] = "ERROR"
606 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
608 vdur
["status-detailed"] = str(error_text
)
610 "vdur.{}.status-detailed".format(vdu_index
)
612 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
613 except DbException
as e
:
614 self
.logger
.error("Cannot update vnf. {}".format(e
))
616 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
618 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
619 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
620 :param nsr_desc_RO: nsr descriptor from RO
621 :return: Nothing, LcmException is raised on errors
623 for vnf_index
, db_vnfr
in db_vnfrs
.items():
624 for vnf_RO
in nsr_desc_RO
["vnfs"]:
625 if vnf_RO
["member_vnf_index"] != vnf_index
:
628 if vnf_RO
.get("ip_address"):
629 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
[
632 elif not db_vnfr
.get("ip-address"):
633 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
634 raise LcmExceptionNoMgmtIP(
635 "ns member_vnf_index '{}' has no IP address".format(
640 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
641 vdur_RO_count_index
= 0
642 if vdur
.get("pdu-type"):
644 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
645 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
647 if vdur
["count-index"] != vdur_RO_count_index
:
648 vdur_RO_count_index
+= 1
650 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
651 if vdur_RO
.get("ip_address"):
652 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
654 vdur
["ip-address"] = None
655 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
656 vdur
["name"] = vdur_RO
.get("vim_name")
657 vdur
["status"] = vdur_RO
.get("status")
658 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
659 for ifacer
in get_iterable(vdur
, "interfaces"):
660 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
661 if ifacer
["name"] == interface_RO
.get("internal_name"):
662 ifacer
["ip-address"] = interface_RO
.get(
665 ifacer
["mac-address"] = interface_RO
.get(
671 "ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
672 "from VIM info".format(
673 vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]
676 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
680 "ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
682 vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]
686 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
687 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
688 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
690 vld
["vim-id"] = net_RO
.get("vim_net_id")
691 vld
["name"] = net_RO
.get("vim_name")
692 vld
["status"] = net_RO
.get("status")
693 vld
["status-detailed"] = net_RO
.get("error_msg")
694 vnfr_update
["vld.{}".format(vld_index
)] = vld
698 "ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
703 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
708 "ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(
713 def _get_ns_config_info(self
, nsr_id
):
715 Generates a mapping between vnf,vdu elements and the N2VC id
716 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
717 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
718 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
719 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
721 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
722 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
724 ns_config_info
= {"osm-config-mapping": mapping
}
725 for vca
in vca_deployed_list
:
726 if not vca
["member-vnf-index"]:
728 if not vca
["vdu_id"]:
729 mapping
[vca
["member-vnf-index"]] = vca
["application"]
733 vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"]
735 ] = vca
["application"]
736 return ns_config_info
738 async def _instantiate_ng_ro(
755 def get_vim_account(vim_account_id
):
757 if vim_account_id
in db_vims
:
758 return db_vims
[vim_account_id
]
759 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account_id
})
760 db_vims
[vim_account_id
] = db_vim
763 # modify target_vld info with instantiation parameters
764 def parse_vld_instantiation_params(
765 target_vim
, target_vld
, vld_params
, target_sdn
767 if vld_params
.get("ip-profile"):
768 target_vld
["vim_info"][target_vim
]["ip_profile"] = vld_params
[
771 if vld_params
.get("provider-network"):
772 target_vld
["vim_info"][target_vim
]["provider_network"] = vld_params
[
775 if "sdn-ports" in vld_params
["provider-network"] and target_sdn
:
776 target_vld
["vim_info"][target_sdn
]["sdn-ports"] = vld_params
[
779 if vld_params
.get("wimAccountId"):
780 target_wim
= "wim:{}".format(vld_params
["wimAccountId"])
781 target_vld
["vim_info"][target_wim
] = {}
782 for param
in ("vim-network-name", "vim-network-id"):
783 if vld_params
.get(param
):
784 if isinstance(vld_params
[param
], dict):
785 for vim
, vim_net
in vld_params
[param
].items():
786 other_target_vim
= "vim:" + vim
788 target_vld
["vim_info"],
789 (other_target_vim
, param
.replace("-", "_")),
792 else: # isinstance str
793 target_vld
["vim_info"][target_vim
][
794 param
.replace("-", "_")
795 ] = vld_params
[param
]
796 if vld_params
.get("common_id"):
797 target_vld
["common_id"] = vld_params
.get("common_id")
799 # modify target["ns"]["vld"] with instantiation parameters to override vnf vim-account
800 def update_ns_vld_target(target
, ns_params
):
801 for vnf_params
in ns_params
.get("vnf", ()):
802 if vnf_params
.get("vimAccountId"):
806 for vnfr
in db_vnfrs
.values()
807 if vnf_params
["member-vnf-index"]
808 == vnfr
["member-vnf-index-ref"]
812 vdur
= next((vdur
for vdur
in target_vnf
.get("vdur", ())), None)
813 for a_index
, a_vld
in enumerate(target
["ns"]["vld"]):
814 target_vld
= find_in_list(
815 get_iterable(vdur
, "interfaces"),
816 lambda iface
: iface
.get("ns-vld-id") == a_vld
["name"],
819 if vnf_params
.get("vimAccountId") not in a_vld
.get(
822 target
["ns"]["vld"][a_index
].get("vim_info").update(
824 "vim:{}".format(vnf_params
["vimAccountId"]): {
825 "vim_network_name": ""
830 nslcmop_id
= db_nslcmop
["_id"]
832 "name": db_nsr
["name"],
835 "image": deepcopy(db_nsr
["image"]),
836 "flavor": deepcopy(db_nsr
["flavor"]),
837 "action_id": nslcmop_id
,
838 "cloud_init_content": {},
840 for image
in target
["image"]:
841 image
["vim_info"] = {}
842 for flavor
in target
["flavor"]:
843 flavor
["vim_info"] = {}
845 if db_nslcmop
.get("lcmOperationType") != "instantiate":
846 # get parameters of instantiation:
847 db_nslcmop_instantiate
= self
.db
.get_list(
850 "nsInstanceId": db_nslcmop
["nsInstanceId"],
851 "lcmOperationType": "instantiate",
854 ns_params
= db_nslcmop_instantiate
.get("operationParams")
856 ns_params
= db_nslcmop
.get("operationParams")
857 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
858 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
861 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
862 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
866 "mgmt-network": vld
.get("mgmt-network", False),
867 "type": vld
.get("type"),
870 "vim_network_name": vld
.get("vim-network-name"),
871 "vim_account_id": ns_params
["vimAccountId"],
875 # check if this network needs SDN assist
876 if vld
.get("pci-interfaces"):
877 db_vim
= get_vim_account(ns_params
["vimAccountId"])
878 sdnc_id
= db_vim
["config"].get("sdn-controller")
880 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
881 target_sdn
= "sdn:{}".format(sdnc_id
)
882 target_vld
["vim_info"][target_sdn
] = {
884 "target_vim": target_vim
,
886 "type": vld
.get("type"),
889 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
890 for nsd_vnf_profile
in nsd_vnf_profiles
:
891 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
892 if cp
["virtual-link-profile-id"] == vld
["id"]:
894 "member_vnf:{}.{}".format(
895 cp
["constituent-cpd-id"][0][
896 "constituent-base-element-id"
898 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
900 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
902 # check at nsd descriptor, if there is an ip-profile
904 nsd_vlp
= find_in_list(
905 get_virtual_link_profiles(nsd
),
906 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
911 and nsd_vlp
.get("virtual-link-protocol-data")
912 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
914 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
917 ip_profile_dest_data
= {}
918 if "ip-version" in ip_profile_source_data
:
919 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
922 if "cidr" in ip_profile_source_data
:
923 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
926 if "gateway-ip" in ip_profile_source_data
:
927 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
930 if "dhcp-enabled" in ip_profile_source_data
:
931 ip_profile_dest_data
["dhcp-params"] = {
932 "enabled": ip_profile_source_data
["dhcp-enabled"]
934 vld_params
["ip-profile"] = ip_profile_dest_data
936 # update vld_params with instantiation params
937 vld_instantiation_params
= find_in_list(
938 get_iterable(ns_params
, "vld"),
939 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
941 if vld_instantiation_params
:
942 vld_params
.update(vld_instantiation_params
)
943 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
944 target
["ns"]["vld"].append(target_vld
)
945 # Update the target ns_vld if vnf vim_account is overriden by instantiation params
946 update_ns_vld_target(target
, ns_params
)
948 for vnfr
in db_vnfrs
.values():
950 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
952 vnf_params
= find_in_list(
953 get_iterable(ns_params
, "vnf"),
954 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
956 target_vnf
= deepcopy(vnfr
)
957 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
958 for vld
in target_vnf
.get("vld", ()):
959 # check if connected to a ns.vld, to fill target'
960 vnf_cp
= find_in_list(
961 vnfd
.get("int-virtual-link-desc", ()),
962 lambda cpd
: cpd
.get("id") == vld
["id"],
965 ns_cp
= "member_vnf:{}.{}".format(
966 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
968 if cp2target
.get(ns_cp
):
969 vld
["target"] = cp2target
[ns_cp
]
972 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
974 # check if this network needs SDN assist
976 if vld
.get("pci-interfaces"):
977 db_vim
= get_vim_account(vnfr
["vim-account-id"])
978 sdnc_id
= db_vim
["config"].get("sdn-controller")
980 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
981 target_sdn
= "sdn:{}".format(sdnc_id
)
982 vld
["vim_info"][target_sdn
] = {
984 "target_vim": target_vim
,
986 "type": vld
.get("type"),
989 # check at vnfd descriptor, if there is an ip-profile
991 vnfd_vlp
= find_in_list(
992 get_virtual_link_profiles(vnfd
),
993 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
997 and vnfd_vlp
.get("virtual-link-protocol-data")
998 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
1000 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
1003 ip_profile_dest_data
= {}
1004 if "ip-version" in ip_profile_source_data
:
1005 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
1008 if "cidr" in ip_profile_source_data
:
1009 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
1012 if "gateway-ip" in ip_profile_source_data
:
1013 ip_profile_dest_data
[
1015 ] = ip_profile_source_data
["gateway-ip"]
1016 if "dhcp-enabled" in ip_profile_source_data
:
1017 ip_profile_dest_data
["dhcp-params"] = {
1018 "enabled": ip_profile_source_data
["dhcp-enabled"]
1021 vld_params
["ip-profile"] = ip_profile_dest_data
1022 # update vld_params with instantiation params
1024 vld_instantiation_params
= find_in_list(
1025 get_iterable(vnf_params
, "internal-vld"),
1026 lambda i_vld
: i_vld
["name"] == vld
["id"],
1028 if vld_instantiation_params
:
1029 vld_params
.update(vld_instantiation_params
)
1030 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1033 for vdur
in target_vnf
.get("vdur", ()):
1034 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1035 continue # This vdu must not be created
1036 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1038 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1041 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1042 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1045 and vdu_configuration
.get("config-access")
1046 and vdu_configuration
.get("config-access").get("ssh-access")
1048 vdur
["ssh-keys"] = ssh_keys_all
1049 vdur
["ssh-access-required"] = vdu_configuration
[
1051 ]["ssh-access"]["required"]
1054 and vnf_configuration
.get("config-access")
1055 and vnf_configuration
.get("config-access").get("ssh-access")
1056 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1058 vdur
["ssh-keys"] = ssh_keys_all
1059 vdur
["ssh-access-required"] = vnf_configuration
[
1061 ]["ssh-access"]["required"]
1062 elif ssh_keys_instantiation
and find_in_list(
1063 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1065 vdur
["ssh-keys"] = ssh_keys_instantiation
1067 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1069 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1071 if vdud
.get("cloud-init-file"):
1072 vdur
["cloud-init"] = "{}:file:{}".format(
1073 vnfd
["_id"], vdud
.get("cloud-init-file")
1075 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1076 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1077 base_folder
= vnfd
["_admin"]["storage"]
1078 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1079 base_folder
["folder"],
1080 base_folder
["pkg-dir"],
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 artifact_path
= "{}/{}/{}/{}".format(
1665 base_folder
["folder"],
1666 base_folder
["pkg-dir"],
1668 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1673 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1675 # get initial_config_primitive_list that applies to this element
1676 initial_config_primitive_list
= config_descriptor
.get(
1677 "initial-config-primitive"
1681 "Initial config primitive list > {}".format(
1682 initial_config_primitive_list
1686 # add config if not present for NS charm
1687 ee_descriptor_id
= ee_config_descriptor
.get("id")
1688 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1689 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1690 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1694 "Initial config primitive list #2 > {}".format(
1695 initial_config_primitive_list
1698 # n2vc_redesign STEP 3.1
1699 # find old ee_id if exists
1700 ee_id
= vca_deployed
.get("ee_id")
1702 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1703 # create or register execution environment in VCA
1704 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1706 self
._write
_configuration
_status
(
1708 vca_index
=vca_index
,
1710 element_under_configuration
=element_under_configuration
,
1711 element_type
=element_type
,
1714 step
= "create execution environment"
1715 self
.logger
.debug(logging_text
+ step
)
1719 if vca_type
== "k8s_proxy_charm":
1720 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1721 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1722 namespace
=namespace
,
1723 artifact_path
=artifact_path
,
1727 elif vca_type
== "helm" or vca_type
== "helm-v3":
1728 ee_id
, credentials
= await self
.vca_map
[
1730 ].create_execution_environment(
1731 namespace
=namespace
,
1735 artifact_path
=artifact_path
,
1739 ee_id
, credentials
= await self
.vca_map
[
1741 ].create_execution_environment(
1742 namespace
=namespace
,
1748 elif vca_type
== "native_charm":
1749 step
= "Waiting to VM being up and getting IP address"
1750 self
.logger
.debug(logging_text
+ step
)
1751 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1760 credentials
= {"hostname": rw_mgmt_ip
}
1762 username
= deep_get(
1763 config_descriptor
, ("config-access", "ssh-access", "default-user")
1765 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1766 # merged. Meanwhile let's get username from initial-config-primitive
1767 if not username
and initial_config_primitive_list
:
1768 for config_primitive
in initial_config_primitive_list
:
1769 for param
in config_primitive
.get("parameter", ()):
1770 if param
["name"] == "ssh-username":
1771 username
= param
["value"]
1775 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1776 "'config-access.ssh-access.default-user'"
1778 credentials
["username"] = username
1779 # n2vc_redesign STEP 3.2
1781 self
._write
_configuration
_status
(
1783 vca_index
=vca_index
,
1784 status
="REGISTERING",
1785 element_under_configuration
=element_under_configuration
,
1786 element_type
=element_type
,
1789 step
= "register execution environment {}".format(credentials
)
1790 self
.logger
.debug(logging_text
+ step
)
1791 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1792 credentials
=credentials
,
1793 namespace
=namespace
,
1798 # for compatibility with MON/POL modules, the need model and application name at database
1799 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1800 ee_id_parts
= ee_id
.split(".")
1801 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1802 if len(ee_id_parts
) >= 2:
1803 model_name
= ee_id_parts
[0]
1804 application_name
= ee_id_parts
[1]
1805 db_nsr_update
[db_update_entry
+ "model"] = model_name
1806 db_nsr_update
[db_update_entry
+ "application"] = application_name
1808 # n2vc_redesign STEP 3.3
1809 step
= "Install configuration Software"
1811 self
._write
_configuration
_status
(
1813 vca_index
=vca_index
,
1814 status
="INSTALLING SW",
1815 element_under_configuration
=element_under_configuration
,
1816 element_type
=element_type
,
1817 other_update
=db_nsr_update
,
1820 # TODO check if already done
1821 self
.logger
.debug(logging_text
+ step
)
1823 if vca_type
== "native_charm":
1824 config_primitive
= next(
1825 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1828 if config_primitive
:
1829 config
= self
._map
_primitive
_params
(
1830 config_primitive
, {}, deploy_params
1833 if vca_type
== "lxc_proxy_charm":
1834 if element_type
== "NS":
1835 num_units
= db_nsr
.get("config-units") or 1
1836 elif element_type
== "VNF":
1837 num_units
= db_vnfr
.get("config-units") or 1
1838 elif element_type
== "VDU":
1839 for v
in db_vnfr
["vdur"]:
1840 if vdu_id
== v
["vdu-id-ref"]:
1841 num_units
= v
.get("config-units") or 1
1843 if vca_type
!= "k8s_proxy_charm":
1844 await self
.vca_map
[vca_type
].install_configuration_sw(
1846 artifact_path
=artifact_path
,
1849 num_units
=num_units
,
1854 # write in db flag of configuration_sw already installed
1856 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1859 # add relations for this VCA (wait for other peers related with this VCA)
1860 await self
._add
_vca
_relations
(
1861 logging_text
=logging_text
,
1863 vca_index
=vca_index
,
1868 # if SSH access is required, then get execution environment SSH public
1869 # if native charm we have waited already to VM be UP
1870 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1873 # self.logger.debug("get ssh key block")
1875 config_descriptor
, ("config-access", "ssh-access", "required")
1877 # self.logger.debug("ssh key needed")
1878 # Needed to inject a ssh key
1881 ("config-access", "ssh-access", "default-user"),
1883 step
= "Install configuration Software, getting public ssh key"
1884 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1885 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1888 step
= "Insert public key into VM user={} ssh_key={}".format(
1892 # self.logger.debug("no need to get ssh key")
1893 step
= "Waiting to VM being up and getting IP address"
1894 self
.logger
.debug(logging_text
+ step
)
1896 # n2vc_redesign STEP 5.1
1897 # wait for RO (ip-address) Insert pub_key into VM
1900 rw_mgmt_ip
= await self
.wait_kdu_up(
1901 logging_text
, nsr_id
, vnfr_id
, kdu_name
1904 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1914 rw_mgmt_ip
= None # This is for a NS configuration
1916 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1918 # store rw_mgmt_ip in deploy params for later replacement
1919 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1921 # n2vc_redesign STEP 6 Execute initial config primitive
1922 step
= "execute initial config primitive"
1924 # wait for dependent primitives execution (NS -> VNF -> VDU)
1925 if initial_config_primitive_list
:
1926 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1928 # stage, in function of element type: vdu, kdu, vnf or ns
1929 my_vca
= vca_deployed_list
[vca_index
]
1930 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1932 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1933 elif my_vca
.get("member-vnf-index"):
1935 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1938 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1940 self
._write
_configuration
_status
(
1941 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1944 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1946 check_if_terminated_needed
= True
1947 for initial_config_primitive
in initial_config_primitive_list
:
1948 # adding information on the vca_deployed if it is a NS execution environment
1949 if not vca_deployed
["member-vnf-index"]:
1950 deploy_params
["ns_config_info"] = json
.dumps(
1951 self
._get
_ns
_config
_info
(nsr_id
)
1953 # TODO check if already done
1954 primitive_params_
= self
._map
_primitive
_params
(
1955 initial_config_primitive
, {}, deploy_params
1958 step
= "execute primitive '{}' params '{}'".format(
1959 initial_config_primitive
["name"], primitive_params_
1961 self
.logger
.debug(logging_text
+ step
)
1962 await self
.vca_map
[vca_type
].exec_primitive(
1964 primitive_name
=initial_config_primitive
["name"],
1965 params_dict
=primitive_params_
,
1970 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1971 if check_if_terminated_needed
:
1972 if config_descriptor
.get("terminate-config-primitive"):
1974 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
1976 check_if_terminated_needed
= False
1978 # TODO register in database that primitive is done
1980 # STEP 7 Configure metrics
1981 if vca_type
== "helm" or vca_type
== "helm-v3":
1982 prometheus_jobs
= await self
.add_prometheus_metrics(
1984 artifact_path
=artifact_path
,
1985 ee_config_descriptor
=ee_config_descriptor
,
1988 target_ip
=rw_mgmt_ip
,
1994 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
1997 step
= "instantiated at VCA"
1998 self
.logger
.debug(logging_text
+ step
)
2000 self
._write
_configuration
_status
(
2001 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
2004 except Exception as e
: # TODO not use Exception but N2VC exception
2005 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
2007 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
2010 "Exception while {} : {}".format(step
, e
), exc_info
=True
2012 self
._write
_configuration
_status
(
2013 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
2015 raise LcmException("{} {}".format(step
, e
)) from e
2017 def _write_ns_status(
2021 current_operation
: str,
2022 current_operation_id
: str,
2023 error_description
: str = None,
2024 error_detail
: str = None,
2025 other_update
: dict = None,
2028 Update db_nsr fields.
2031 :param current_operation:
2032 :param current_operation_id:
2033 :param error_description:
2034 :param error_detail:
2035 :param other_update: Other required changes at database if provided, will be cleared
2039 db_dict
= other_update
or {}
2042 ] = current_operation_id
# for backward compatibility
2043 db_dict
["_admin.current-operation"] = current_operation_id
2044 db_dict
["_admin.operation-type"] = (
2045 current_operation
if current_operation
!= "IDLE" else None
2047 db_dict
["currentOperation"] = current_operation
2048 db_dict
["currentOperationID"] = current_operation_id
2049 db_dict
["errorDescription"] = error_description
2050 db_dict
["errorDetail"] = error_detail
2053 db_dict
["nsState"] = ns_state
2054 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2055 except DbException
as e
:
2056 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2058 def _write_op_status(
2062 error_message
: str = None,
2063 queuePosition
: int = 0,
2064 operation_state
: str = None,
2065 other_update
: dict = None,
2068 db_dict
= other_update
or {}
2069 db_dict
["queuePosition"] = queuePosition
2070 if isinstance(stage
, list):
2071 db_dict
["stage"] = stage
[0]
2072 db_dict
["detailed-status"] = " ".join(stage
)
2073 elif stage
is not None:
2074 db_dict
["stage"] = str(stage
)
2076 if error_message
is not None:
2077 db_dict
["errorMessage"] = error_message
2078 if operation_state
is not None:
2079 db_dict
["operationState"] = operation_state
2080 db_dict
["statusEnteredTime"] = time()
2081 self
.update_db_2("nslcmops", op_id
, db_dict
)
2082 except DbException
as e
:
2084 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2087 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2089 nsr_id
= db_nsr
["_id"]
2090 # configurationStatus
2091 config_status
= db_nsr
.get("configurationStatus")
2094 "configurationStatus.{}.status".format(index
): status
2095 for index
, v
in enumerate(config_status
)
2099 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2101 except DbException
as e
:
2103 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2106 def _write_configuration_status(
2111 element_under_configuration
: str = None,
2112 element_type
: str = None,
2113 other_update
: dict = None,
2116 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2117 # .format(vca_index, status))
2120 db_path
= "configurationStatus.{}.".format(vca_index
)
2121 db_dict
= other_update
or {}
2123 db_dict
[db_path
+ "status"] = status
2124 if element_under_configuration
:
2126 db_path
+ "elementUnderConfiguration"
2127 ] = element_under_configuration
2129 db_dict
[db_path
+ "elementType"] = element_type
2130 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2131 except DbException
as e
:
2133 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2134 status
, nsr_id
, vca_index
, e
2138 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2140 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2141 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2142 Database is used because the result can be obtained from a different LCM worker in case of HA.
2143 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2144 :param db_nslcmop: database content of nslcmop
2145 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2146 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2147 computed 'vim-account-id'
2150 nslcmop_id
= db_nslcmop
["_id"]
2151 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2152 if placement_engine
== "PLA":
2154 logging_text
+ "Invoke and wait for placement optimization"
2156 await self
.msg
.aiowrite(
2157 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2159 db_poll_interval
= 5
2160 wait
= db_poll_interval
* 10
2162 while not pla_result
and wait
>= 0:
2163 await asyncio
.sleep(db_poll_interval
)
2164 wait
-= db_poll_interval
2165 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2166 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2170 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2173 for pla_vnf
in pla_result
["vnf"]:
2174 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2175 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2180 {"_id": vnfr
["_id"]},
2181 {"vim-account-id": pla_vnf
["vimAccountId"]},
2184 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2187 def update_nsrs_with_pla_result(self
, params
):
2189 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2191 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2193 except Exception as e
:
2194 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2196 async def instantiate(self
, nsr_id
, nslcmop_id
):
2199 :param nsr_id: ns instance to deploy
2200 :param nslcmop_id: operation to run
2204 # Try to lock HA task here
2205 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2206 if not task_is_locked_by_me
:
2208 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2212 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2213 self
.logger
.debug(logging_text
+ "Enter")
2215 # get all needed from database
2217 # database nsrs record
2220 # database nslcmops record
2223 # update operation on nsrs
2225 # update operation on nslcmops
2226 db_nslcmop_update
= {}
2228 nslcmop_operation_state
= None
2229 db_vnfrs
= {} # vnf's info indexed by member-index
2231 tasks_dict_info
= {} # from task to info text
2235 "Stage 1/5: preparation of the environment.",
2236 "Waiting for previous operations to terminate.",
2239 # ^ stage, step, VIM progress
2241 # wait for any previous tasks in process
2242 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2244 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2245 stage
[1] = "Reading from database."
2246 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2247 db_nsr_update
["detailed-status"] = "creating"
2248 db_nsr_update
["operational-status"] = "init"
2249 self
._write
_ns
_status
(
2251 ns_state
="BUILDING",
2252 current_operation
="INSTANTIATING",
2253 current_operation_id
=nslcmop_id
,
2254 other_update
=db_nsr_update
,
2256 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2258 # read from db: operation
2259 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2260 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2261 if db_nslcmop
["operationParams"].get("additionalParamsForVnf"):
2262 db_nslcmop
["operationParams"]["additionalParamsForVnf"] = json
.loads(
2263 db_nslcmop
["operationParams"]["additionalParamsForVnf"]
2265 ns_params
= db_nslcmop
.get("operationParams")
2266 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2267 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2269 timeout_ns_deploy
= self
.timeout
.get(
2270 "ns_deploy", self
.timeout_ns_deploy
2274 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2275 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2276 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2277 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2278 self
.fs
.sync(db_nsr
["nsd-id"])
2280 # nsr_name = db_nsr["name"] # TODO short-name??
2282 # read from db: vnf's of this ns
2283 stage
[1] = "Getting vnfrs from db."
2284 self
.logger
.debug(logging_text
+ stage
[1])
2285 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2287 # read from db: vnfd's for every vnf
2288 db_vnfds
= [] # every vnfd data
2290 # for each vnf in ns, read vnfd
2291 for vnfr
in db_vnfrs_list
:
2292 if vnfr
.get("kdur"):
2294 for kdur
in vnfr
["kdur"]:
2295 if kdur
.get("additionalParams"):
2296 kdur
["additionalParams"] = json
.loads(
2297 kdur
["additionalParams"]
2299 kdur_list
.append(kdur
)
2300 vnfr
["kdur"] = kdur_list
2302 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2303 vnfd_id
= vnfr
["vnfd-id"]
2304 vnfd_ref
= vnfr
["vnfd-ref"]
2305 self
.fs
.sync(vnfd_id
)
2307 # if we haven't this vnfd, read it from db
2308 if vnfd_id
not in db_vnfds
:
2310 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2313 self
.logger
.debug(logging_text
+ stage
[1])
2314 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2317 db_vnfds
.append(vnfd
)
2319 # Get or generates the _admin.deployed.VCA list
2320 vca_deployed_list
= None
2321 if db_nsr
["_admin"].get("deployed"):
2322 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2323 if vca_deployed_list
is None:
2324 vca_deployed_list
= []
2325 configuration_status_list
= []
2326 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2327 db_nsr_update
["configurationStatus"] = configuration_status_list
2328 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2329 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2330 elif isinstance(vca_deployed_list
, dict):
2331 # maintain backward compatibility. Change a dict to list at database
2332 vca_deployed_list
= list(vca_deployed_list
.values())
2333 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2334 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2337 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2339 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2340 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2342 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2343 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2344 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2346 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2349 # n2vc_redesign STEP 2 Deploy Network Scenario
2350 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2351 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2353 stage
[1] = "Deploying KDUs."
2354 # self.logger.debug(logging_text + "Before deploy_kdus")
2355 # Call to deploy_kdus in case exists the "vdu:kdu" param
2356 await self
.deploy_kdus(
2357 logging_text
=logging_text
,
2359 nslcmop_id
=nslcmop_id
,
2362 task_instantiation_info
=tasks_dict_info
,
2365 stage
[1] = "Getting VCA public key."
2366 # n2vc_redesign STEP 1 Get VCA public ssh-key
2367 # feature 1429. Add n2vc public key to needed VMs
2368 n2vc_key
= self
.n2vc
.get_public_key()
2369 n2vc_key_list
= [n2vc_key
]
2370 if self
.vca_config
.get("public_key"):
2371 n2vc_key_list
.append(self
.vca_config
["public_key"])
2373 stage
[1] = "Deploying NS at VIM."
2374 task_ro
= asyncio
.ensure_future(
2375 self
.instantiate_RO(
2376 logging_text
=logging_text
,
2380 db_nslcmop
=db_nslcmop
,
2383 n2vc_key_list
=n2vc_key_list
,
2387 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2388 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2390 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2391 stage
[1] = "Deploying Execution Environments."
2392 self
.logger
.debug(logging_text
+ stage
[1])
2394 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2395 for vnf_profile
in get_vnf_profiles(nsd
):
2396 vnfd_id
= vnf_profile
["vnfd-id"]
2397 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2398 member_vnf_index
= str(vnf_profile
["id"])
2399 db_vnfr
= db_vnfrs
[member_vnf_index
]
2400 base_folder
= vnfd
["_admin"]["storage"]
2406 # Get additional parameters
2407 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2408 if db_vnfr
.get("additionalParamsForVnf"):
2409 deploy_params
.update(
2410 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2413 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2414 if descriptor_config
:
2416 logging_text
=logging_text
2417 + "member_vnf_index={} ".format(member_vnf_index
),
2420 nslcmop_id
=nslcmop_id
,
2426 member_vnf_index
=member_vnf_index
,
2427 vdu_index
=vdu_index
,
2429 deploy_params
=deploy_params
,
2430 descriptor_config
=descriptor_config
,
2431 base_folder
=base_folder
,
2432 task_instantiation_info
=tasks_dict_info
,
2436 # Deploy charms for each VDU that supports one.
2437 for vdud
in get_vdu_list(vnfd
):
2439 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2440 vdur
= find_in_list(
2441 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2444 if vdur
.get("additionalParams"):
2445 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2447 deploy_params_vdu
= deploy_params
2448 deploy_params_vdu
["OSM"] = get_osm_params(
2449 db_vnfr
, vdu_id
, vdu_count_index
=0
2451 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2453 self
.logger
.debug("VDUD > {}".format(vdud
))
2455 "Descriptor config > {}".format(descriptor_config
)
2457 if descriptor_config
:
2460 for vdu_index
in range(vdud_count
):
2461 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2463 logging_text
=logging_text
2464 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2465 member_vnf_index
, vdu_id
, vdu_index
2469 nslcmop_id
=nslcmop_id
,
2475 member_vnf_index
=member_vnf_index
,
2476 vdu_index
=vdu_index
,
2478 deploy_params
=deploy_params_vdu
,
2479 descriptor_config
=descriptor_config
,
2480 base_folder
=base_folder
,
2481 task_instantiation_info
=tasks_dict_info
,
2484 for kdud
in get_kdu_list(vnfd
):
2485 kdu_name
= kdud
["name"]
2486 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2487 if descriptor_config
:
2492 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2494 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2495 if kdur
.get("additionalParams"):
2496 deploy_params_kdu
.update(
2497 parse_yaml_strings(kdur
["additionalParams"].copy())
2501 logging_text
=logging_text
,
2504 nslcmop_id
=nslcmop_id
,
2510 member_vnf_index
=member_vnf_index
,
2511 vdu_index
=vdu_index
,
2513 deploy_params
=deploy_params_kdu
,
2514 descriptor_config
=descriptor_config
,
2515 base_folder
=base_folder
,
2516 task_instantiation_info
=tasks_dict_info
,
2520 # Check if this NS has a charm configuration
2521 descriptor_config
= nsd
.get("ns-configuration")
2522 if descriptor_config
and descriptor_config
.get("juju"):
2525 member_vnf_index
= None
2531 # Get additional parameters
2532 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2533 if db_nsr
.get("additionalParamsForNs"):
2534 deploy_params
.update(
2535 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2537 base_folder
= nsd
["_admin"]["storage"]
2539 logging_text
=logging_text
,
2542 nslcmop_id
=nslcmop_id
,
2548 member_vnf_index
=member_vnf_index
,
2549 vdu_index
=vdu_index
,
2551 deploy_params
=deploy_params
,
2552 descriptor_config
=descriptor_config
,
2553 base_folder
=base_folder
,
2554 task_instantiation_info
=tasks_dict_info
,
2558 # rest of staff will be done at finally
2561 ROclient
.ROClientException
,
2567 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2570 except asyncio
.CancelledError
:
2572 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2574 exc
= "Operation was cancelled"
2575 except Exception as e
:
2576 exc
= traceback
.format_exc()
2577 self
.logger
.critical(
2578 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2583 error_list
.append(str(exc
))
2585 # wait for pending tasks
2587 stage
[1] = "Waiting for instantiate pending tasks."
2588 self
.logger
.debug(logging_text
+ stage
[1])
2589 error_list
+= await self
._wait
_for
_tasks
(
2597 stage
[1] = stage
[2] = ""
2598 except asyncio
.CancelledError
:
2599 error_list
.append("Cancelled")
2600 # TODO cancel all tasks
2601 except Exception as exc
:
2602 error_list
.append(str(exc
))
2604 # update operation-status
2605 db_nsr_update
["operational-status"] = "running"
2606 # let's begin with VCA 'configured' status (later we can change it)
2607 db_nsr_update
["config-status"] = "configured"
2608 for task
, task_name
in tasks_dict_info
.items():
2609 if not task
.done() or task
.cancelled() or task
.exception():
2610 if task_name
.startswith(self
.task_name_deploy_vca
):
2611 # A N2VC task is pending
2612 db_nsr_update
["config-status"] = "failed"
2614 # RO or KDU task is pending
2615 db_nsr_update
["operational-status"] = "failed"
2617 # update status at database
2619 error_detail
= ". ".join(error_list
)
2620 self
.logger
.error(logging_text
+ error_detail
)
2621 error_description_nslcmop
= "{} Detail: {}".format(
2622 stage
[0], error_detail
2624 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2625 nslcmop_id
, stage
[0]
2628 db_nsr_update
["detailed-status"] = (
2629 error_description_nsr
+ " Detail: " + error_detail
2631 db_nslcmop_update
["detailed-status"] = error_detail
2632 nslcmop_operation_state
= "FAILED"
2636 error_description_nsr
= error_description_nslcmop
= None
2638 db_nsr_update
["detailed-status"] = "Done"
2639 db_nslcmop_update
["detailed-status"] = "Done"
2640 nslcmop_operation_state
= "COMPLETED"
2643 self
._write
_ns
_status
(
2646 current_operation
="IDLE",
2647 current_operation_id
=None,
2648 error_description
=error_description_nsr
,
2649 error_detail
=error_detail
,
2650 other_update
=db_nsr_update
,
2652 self
._write
_op
_status
(
2655 error_message
=error_description_nslcmop
,
2656 operation_state
=nslcmop_operation_state
,
2657 other_update
=db_nslcmop_update
,
2660 if nslcmop_operation_state
:
2662 await self
.msg
.aiowrite(
2667 "nslcmop_id": nslcmop_id
,
2668 "operationState": nslcmop_operation_state
,
2672 except Exception as e
:
2674 logging_text
+ "kafka_write notification Exception {}".format(e
)
2677 self
.logger
.debug(logging_text
+ "Exit")
2678 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2680 async def _add_vca_relations(
2685 timeout
: int = 3600,
2686 vca_type
: str = None,
2691 # 1. find all relations for this VCA
2692 # 2. wait for other peers related
2696 vca_type
= vca_type
or "lxc_proxy_charm"
2698 # STEP 1: find all relations for this VCA
2701 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2702 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2705 my_vca
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))[vca_index
]
2707 # read all ns-configuration relations
2708 ns_relations
= list()
2709 db_ns_relations
= deep_get(nsd
, ("ns-configuration", "relation"))
2711 for r
in db_ns_relations
:
2712 # check if this VCA is in the relation
2713 if my_vca
.get("member-vnf-index") in (
2714 r
.get("entities")[0].get("id"),
2715 r
.get("entities")[1].get("id"),
2717 ns_relations
.append(r
)
2719 # read all vnf-configuration relations
2720 vnf_relations
= list()
2721 db_vnfd_list
= db_nsr
.get("vnfd-id")
2723 for vnfd
in db_vnfd_list
:
2724 db_vnf_relations
= None
2725 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2726 db_vnf_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
2727 if db_vnf_configuration
:
2728 db_vnf_relations
= db_vnf_configuration
.get("relation", [])
2729 if db_vnf_relations
:
2730 for r
in db_vnf_relations
:
2731 # check if this VCA is in the relation
2732 if my_vca
.get("vdu_id") in (
2733 r
.get("entities")[0].get("id"),
2734 r
.get("entities")[1].get("id"),
2736 vnf_relations
.append(r
)
2738 # if no relations, terminate
2739 if not ns_relations
and not vnf_relations
:
2740 self
.logger
.debug(logging_text
+ " No relations")
2745 + " adding relations\n {}\n {}".format(
2746 ns_relations
, vnf_relations
2755 if now
- start
>= timeout
:
2756 self
.logger
.error(logging_text
+ " : timeout adding relations")
2759 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2760 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2762 # for each defined NS relation, find the VCA's related
2763 for r
in ns_relations
.copy():
2764 from_vca_ee_id
= None
2766 from_vca_endpoint
= None
2767 to_vca_endpoint
= None
2768 vca_list
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))
2769 for vca
in vca_list
:
2770 if vca
.get("member-vnf-index") == r
.get("entities")[0].get(
2772 ) and vca
.get("config_sw_installed"):
2773 from_vca_ee_id
= vca
.get("ee_id")
2774 from_vca_endpoint
= r
.get("entities")[0].get("endpoint")
2775 if vca
.get("member-vnf-index") == r
.get("entities")[1].get(
2777 ) and vca
.get("config_sw_installed"):
2778 to_vca_ee_id
= vca
.get("ee_id")
2779 to_vca_endpoint
= r
.get("entities")[1].get("endpoint")
2780 if from_vca_ee_id
and to_vca_ee_id
:
2782 await self
.vca_map
[vca_type
].add_relation(
2783 ee_id_1
=from_vca_ee_id
,
2784 ee_id_2
=to_vca_ee_id
,
2785 endpoint_1
=from_vca_endpoint
,
2786 endpoint_2
=to_vca_endpoint
,
2789 # remove entry from relations list
2790 ns_relations
.remove(r
)
2792 # check failed peers
2794 vca_status_list
= db_nsr
.get("configurationStatus")
2796 for i
in range(len(vca_list
)):
2798 vca_status
= vca_status_list
[i
]
2799 if vca
.get("member-vnf-index") == r
.get("entities")[
2802 if vca_status
.get("status") == "BROKEN":
2803 # peer broken: remove relation from list
2804 ns_relations
.remove(r
)
2805 if vca
.get("member-vnf-index") == r
.get("entities")[
2808 if vca_status
.get("status") == "BROKEN":
2809 # peer broken: remove relation from list
2810 ns_relations
.remove(r
)
2815 # for each defined VNF relation, find the VCA's related
2816 for r
in vnf_relations
.copy():
2817 from_vca_ee_id
= None
2819 from_vca_endpoint
= None
2820 to_vca_endpoint
= None
2821 vca_list
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))
2822 for vca
in vca_list
:
2823 key_to_check
= "vdu_id"
2824 if vca
.get("vdu_id") is None:
2825 key_to_check
= "vnfd_id"
2826 if vca
.get(key_to_check
) == r
.get("entities")[0].get(
2828 ) and vca
.get("config_sw_installed"):
2829 from_vca_ee_id
= vca
.get("ee_id")
2830 from_vca_endpoint
= r
.get("entities")[0].get("endpoint")
2831 if vca
.get(key_to_check
) == r
.get("entities")[1].get(
2833 ) and vca
.get("config_sw_installed"):
2834 to_vca_ee_id
= vca
.get("ee_id")
2835 to_vca_endpoint
= r
.get("entities")[1].get("endpoint")
2836 if from_vca_ee_id
and to_vca_ee_id
:
2838 await self
.vca_map
[vca_type
].add_relation(
2839 ee_id_1
=from_vca_ee_id
,
2840 ee_id_2
=to_vca_ee_id
,
2841 endpoint_1
=from_vca_endpoint
,
2842 endpoint_2
=to_vca_endpoint
,
2845 # remove entry from relations list
2846 vnf_relations
.remove(r
)
2848 # check failed peers
2850 vca_status_list
= db_nsr
.get("configurationStatus")
2852 for i
in range(len(vca_list
)):
2854 vca_status
= vca_status_list
[i
]
2855 if vca
.get("vdu_id") == r
.get("entities")[0].get(
2858 if vca_status
.get("status") == "BROKEN":
2859 # peer broken: remove relation from list
2860 vnf_relations
.remove(r
)
2861 if vca
.get("vdu_id") == r
.get("entities")[1].get(
2864 if vca_status
.get("status") == "BROKEN":
2865 # peer broken: remove relation from list
2866 vnf_relations
.remove(r
)
2872 await asyncio
.sleep(5.0)
2874 if not ns_relations
and not vnf_relations
:
2875 self
.logger
.debug("Relations added")
2880 except Exception as e
:
2881 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
2884 async def _install_kdu(
2892 k8s_instance_info
: dict,
2893 k8params
: dict = None,
2899 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
2902 "collection": "nsrs",
2903 "filter": {"_id": nsr_id
},
2904 "path": nsr_db_path
,
2907 if k8s_instance_info
.get("kdu-deployment-name"):
2908 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
2910 kdu_instance
= self
.k8scluster_map
[
2912 ].generate_kdu_instance_name(
2913 db_dict
=db_dict_install
,
2914 kdu_model
=k8s_instance_info
["kdu-model"],
2915 kdu_name
=k8s_instance_info
["kdu-name"],
2918 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
2920 await self
.k8scluster_map
[k8sclustertype
].install(
2921 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2922 kdu_model
=k8s_instance_info
["kdu-model"],
2925 db_dict
=db_dict_install
,
2927 kdu_name
=k8s_instance_info
["kdu-name"],
2928 namespace
=k8s_instance_info
["namespace"],
2929 kdu_instance
=kdu_instance
,
2933 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
2936 # Obtain services to obtain management service ip
2937 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
2938 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2939 kdu_instance
=kdu_instance
,
2940 namespace
=k8s_instance_info
["namespace"],
2943 # Obtain management service info (if exists)
2944 vnfr_update_dict
= {}
2945 kdu_config
= get_configuration(vnfd
, kdud
["name"])
2947 target_ee_list
= kdu_config
.get("execution-environment-list", [])
2952 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
2955 for service
in kdud
.get("service", [])
2956 if service
.get("mgmt-service")
2958 for mgmt_service
in mgmt_services
:
2959 for service
in services
:
2960 if service
["name"].startswith(mgmt_service
["name"]):
2961 # Mgmt service found, Obtain service ip
2962 ip
= service
.get("external_ip", service
.get("cluster_ip"))
2963 if isinstance(ip
, list) and len(ip
) == 1:
2967 "kdur.{}.ip-address".format(kdu_index
)
2970 # Check if must update also mgmt ip at the vnf
2971 service_external_cp
= mgmt_service
.get(
2972 "external-connection-point-ref"
2974 if service_external_cp
:
2976 deep_get(vnfd
, ("mgmt-interface", "cp"))
2977 == service_external_cp
2979 vnfr_update_dict
["ip-address"] = ip
2984 "external-connection-point-ref", ""
2986 == service_external_cp
,
2989 "kdur.{}.ip-address".format(kdu_index
)
2994 "Mgmt service name: {} not found".format(
2995 mgmt_service
["name"]
2999 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3000 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3002 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3005 and kdu_config
.get("initial-config-primitive")
3006 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3008 initial_config_primitive_list
= kdu_config
.get(
3009 "initial-config-primitive"
3011 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3013 for initial_config_primitive
in initial_config_primitive_list
:
3014 primitive_params_
= self
._map
_primitive
_params
(
3015 initial_config_primitive
, {}, {}
3018 await asyncio
.wait_for(
3019 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3020 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3021 kdu_instance
=kdu_instance
,
3022 primitive_name
=initial_config_primitive
["name"],
3023 params
=primitive_params_
,
3024 db_dict
=db_dict_install
,
3030 except Exception as e
:
3031 # Prepare update db with error and raise exception
3034 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3038 vnfr_data
.get("_id"),
3039 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3042 # ignore to keep original exception
3044 # reraise original error
3049 async def deploy_kdus(
3056 task_instantiation_info
,
3058 # Launch kdus if present in the descriptor
3060 k8scluster_id_2_uuic
= {
3061 "helm-chart-v3": {},
3066 async def _get_cluster_id(cluster_id
, cluster_type
):
3067 nonlocal k8scluster_id_2_uuic
3068 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3069 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3071 # check if K8scluster is creating and wait look if previous tasks in process
3072 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3073 "k8scluster", cluster_id
3076 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3077 task_name
, cluster_id
3079 self
.logger
.debug(logging_text
+ text
)
3080 await asyncio
.wait(task_dependency
, timeout
=3600)
3082 db_k8scluster
= self
.db
.get_one(
3083 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3085 if not db_k8scluster
:
3086 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3088 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3090 if cluster_type
== "helm-chart-v3":
3092 # backward compatibility for existing clusters that have not been initialized for helm v3
3093 k8s_credentials
= yaml
.safe_dump(
3094 db_k8scluster
.get("credentials")
3096 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3097 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3099 db_k8scluster_update
= {}
3100 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3101 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3102 db_k8scluster_update
[
3103 "_admin.helm-chart-v3.created"
3105 db_k8scluster_update
[
3106 "_admin.helm-chart-v3.operationalState"
3109 "k8sclusters", cluster_id
, db_k8scluster_update
3111 except Exception as e
:
3114 + "error initializing helm-v3 cluster: {}".format(str(e
))
3117 "K8s cluster '{}' has not been initialized for '{}'".format(
3118 cluster_id
, cluster_type
3123 "K8s cluster '{}' has not been initialized for '{}'".format(
3124 cluster_id
, cluster_type
3127 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3130 logging_text
+= "Deploy kdus: "
3133 db_nsr_update
= {"_admin.deployed.K8s": []}
3134 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3137 updated_cluster_list
= []
3138 updated_v3_cluster_list
= []
3140 for vnfr_data
in db_vnfrs
.values():
3141 vca_id
= self
.get_vca_id(vnfr_data
, {})
3142 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3143 # Step 0: Prepare and set parameters
3144 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3145 vnfd_id
= vnfr_data
.get("vnfd-id")
3146 vnfd_with_id
= find_in_list(
3147 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3151 for kdud
in vnfd_with_id
["kdu"]
3152 if kdud
["name"] == kdur
["kdu-name"]
3154 namespace
= kdur
.get("k8s-namespace")
3155 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3156 if kdur
.get("helm-chart"):
3157 kdumodel
= kdur
["helm-chart"]
3158 # Default version: helm3, if helm-version is v2 assign v2
3159 k8sclustertype
= "helm-chart-v3"
3160 self
.logger
.debug("kdur: {}".format(kdur
))
3162 kdur
.get("helm-version")
3163 and kdur
.get("helm-version") == "v2"
3165 k8sclustertype
= "helm-chart"
3166 elif kdur
.get("juju-bundle"):
3167 kdumodel
= kdur
["juju-bundle"]
3168 k8sclustertype
= "juju-bundle"
3171 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3172 "juju-bundle. Maybe an old NBI version is running".format(
3173 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3176 # check if kdumodel is a file and exists
3178 vnfd_with_id
= find_in_list(
3179 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3181 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3182 if storage
and storage
.get(
3184 ): # may be not present if vnfd has not artifacts
3185 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3186 filename
= "{}/{}/{}s/{}".format(
3192 if self
.fs
.file_exists(
3193 filename
, mode
="file"
3194 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3195 kdumodel
= self
.fs
.path
+ filename
3196 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3198 except Exception: # it is not a file
3201 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3202 step
= "Synchronize repos for k8s cluster '{}'".format(
3205 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3209 k8sclustertype
== "helm-chart"
3210 and cluster_uuid
not in updated_cluster_list
3212 k8sclustertype
== "helm-chart-v3"
3213 and cluster_uuid
not in updated_v3_cluster_list
3215 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3216 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3217 cluster_uuid
=cluster_uuid
3220 if del_repo_list
or added_repo_dict
:
3221 if k8sclustertype
== "helm-chart":
3223 "_admin.helm_charts_added." + item
: None
3224 for item
in del_repo_list
3227 "_admin.helm_charts_added." + item
: name
3228 for item
, name
in added_repo_dict
.items()
3230 updated_cluster_list
.append(cluster_uuid
)
3231 elif k8sclustertype
== "helm-chart-v3":
3233 "_admin.helm_charts_v3_added." + item
: None
3234 for item
in del_repo_list
3237 "_admin.helm_charts_v3_added." + item
: name
3238 for item
, name
in added_repo_dict
.items()
3240 updated_v3_cluster_list
.append(cluster_uuid
)
3242 logging_text
+ "repos synchronized on k8s cluster "
3243 "'{}' to_delete: {}, to_add: {}".format(
3244 k8s_cluster_id
, del_repo_list
, added_repo_dict
3249 {"_id": k8s_cluster_id
},
3255 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3256 vnfr_data
["member-vnf-index-ref"],
3260 k8s_instance_info
= {
3261 "kdu-instance": None,
3262 "k8scluster-uuid": cluster_uuid
,
3263 "k8scluster-type": k8sclustertype
,
3264 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3265 "kdu-name": kdur
["kdu-name"],
3266 "kdu-model": kdumodel
,
3267 "namespace": namespace
,
3268 "kdu-deployment-name": kdu_deployment_name
,
3270 db_path
= "_admin.deployed.K8s.{}".format(index
)
3271 db_nsr_update
[db_path
] = k8s_instance_info
3272 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3273 vnfd_with_id
= find_in_list(
3274 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3276 task
= asyncio
.ensure_future(
3285 k8params
=desc_params
,
3290 self
.lcm_tasks
.register(
3294 "instantiate_KDU-{}".format(index
),
3297 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3303 except (LcmException
, asyncio
.CancelledError
):
3305 except Exception as e
:
3306 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3307 if isinstance(e
, (N2VCException
, DbException
)):
3308 self
.logger
.error(logging_text
+ msg
)
3310 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3311 raise LcmException(msg
)
3314 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3333 task_instantiation_info
,
3336 # launch instantiate_N2VC in a asyncio task and register task object
3337 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3338 # if not found, create one entry and update database
3339 # fill db_nsr._admin.deployed.VCA.<index>
3342 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3344 if "execution-environment-list" in descriptor_config
:
3345 ee_list
= descriptor_config
.get("execution-environment-list", [])
3346 elif "juju" in descriptor_config
:
3347 ee_list
= [descriptor_config
] # ns charms
3348 else: # other types as script are not supported
3351 for ee_item
in ee_list
:
3354 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3355 ee_item
.get("juju"), ee_item
.get("helm-chart")
3358 ee_descriptor_id
= ee_item
.get("id")
3359 if ee_item
.get("juju"):
3360 vca_name
= ee_item
["juju"].get("charm")
3363 if ee_item
["juju"].get("charm") is not None
3366 if ee_item
["juju"].get("cloud") == "k8s":
3367 vca_type
= "k8s_proxy_charm"
3368 elif ee_item
["juju"].get("proxy") is False:
3369 vca_type
= "native_charm"
3370 elif ee_item
.get("helm-chart"):
3371 vca_name
= ee_item
["helm-chart"]
3372 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3375 vca_type
= "helm-v3"
3378 logging_text
+ "skipping non juju neither charm configuration"
3383 for vca_index
, vca_deployed
in enumerate(
3384 db_nsr
["_admin"]["deployed"]["VCA"]
3386 if not vca_deployed
:
3389 vca_deployed
.get("member-vnf-index") == member_vnf_index
3390 and vca_deployed
.get("vdu_id") == vdu_id
3391 and vca_deployed
.get("kdu_name") == kdu_name
3392 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3393 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3397 # not found, create one.
3399 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3402 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3404 target
+= "/kdu/{}".format(kdu_name
)
3406 "target_element": target
,
3407 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3408 "member-vnf-index": member_vnf_index
,
3410 "kdu_name": kdu_name
,
3411 "vdu_count_index": vdu_index
,
3412 "operational-status": "init", # TODO revise
3413 "detailed-status": "", # TODO revise
3414 "step": "initial-deploy", # TODO revise
3416 "vdu_name": vdu_name
,
3418 "ee_descriptor_id": ee_descriptor_id
,
3422 # create VCA and configurationStatus in db
3424 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3425 "configurationStatus.{}".format(vca_index
): dict(),
3427 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3429 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3431 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3432 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3433 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3436 task_n2vc
= asyncio
.ensure_future(
3437 self
.instantiate_N2VC(
3438 logging_text
=logging_text
,
3439 vca_index
=vca_index
,
3445 vdu_index
=vdu_index
,
3446 deploy_params
=deploy_params
,
3447 config_descriptor
=descriptor_config
,
3448 base_folder
=base_folder
,
3449 nslcmop_id
=nslcmop_id
,
3453 ee_config_descriptor
=ee_item
,
3456 self
.lcm_tasks
.register(
3460 "instantiate_N2VC-{}".format(vca_index
),
3463 task_instantiation_info
[
3465 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3466 member_vnf_index
or "", vdu_id
or ""
3470 def _create_nslcmop(nsr_id
, operation
, params
):
3472 Creates a ns-lcm-opp content to be stored at database.
3473 :param nsr_id: internal id of the instance
3474 :param operation: instantiate, terminate, scale, action, ...
3475 :param params: user parameters for the operation
3476 :return: dictionary following SOL005 format
3478 # Raise exception if invalid arguments
3479 if not (nsr_id
and operation
and params
):
3481 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3488 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3489 "operationState": "PROCESSING",
3490 "statusEnteredTime": now
,
3491 "nsInstanceId": nsr_id
,
3492 "lcmOperationType": operation
,
3494 "isAutomaticInvocation": False,
3495 "operationParams": params
,
3496 "isCancelPending": False,
3498 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3499 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3504 def _format_additional_params(self
, params
):
3505 params
= params
or {}
3506 for key
, value
in params
.items():
3507 if str(value
).startswith("!!yaml "):
3508 params
[key
] = yaml
.safe_load(value
[7:])
3511 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3512 primitive
= seq
.get("name")
3513 primitive_params
= {}
3515 "member_vnf_index": vnf_index
,
3516 "primitive": primitive
,
3517 "primitive_params": primitive_params
,
3520 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3524 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3525 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3526 if op
.get("operationState") == "COMPLETED":
3527 # b. Skip sub-operation
3528 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3529 return self
.SUBOPERATION_STATUS_SKIP
3531 # c. retry executing sub-operation
3532 # The sub-operation exists, and operationState != 'COMPLETED'
3533 # Update operationState = 'PROCESSING' to indicate a retry.
3534 operationState
= "PROCESSING"
3535 detailed_status
= "In progress"
3536 self
._update
_suboperation
_status
(
3537 db_nslcmop
, op_index
, operationState
, detailed_status
3539 # Return the sub-operation index
3540 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3541 # with arguments extracted from the sub-operation
3544 # Find a sub-operation where all keys in a matching dictionary must match
3545 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3546 def _find_suboperation(self
, db_nslcmop
, match
):
3547 if db_nslcmop
and match
:
3548 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3549 for i
, op
in enumerate(op_list
):
3550 if all(op
.get(k
) == match
[k
] for k
in match
):
3552 return self
.SUBOPERATION_STATUS_NOT_FOUND
3554 # Update status for a sub-operation given its index
3555 def _update_suboperation_status(
3556 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3558 # Update DB for HA tasks
3559 q_filter
= {"_id": db_nslcmop
["_id"]}
3561 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3562 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3565 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3568 # Add sub-operation, return the index of the added sub-operation
3569 # Optionally, set operationState, detailed-status, and operationType
3570 # Status and type are currently set for 'scale' sub-operations:
3571 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3572 # 'detailed-status' : status message
3573 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3574 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3575 def _add_suboperation(
3583 mapped_primitive_params
,
3584 operationState
=None,
3585 detailed_status
=None,
3588 RO_scaling_info
=None,
3591 return self
.SUBOPERATION_STATUS_NOT_FOUND
3592 # Get the "_admin.operations" list, if it exists
3593 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3594 op_list
= db_nslcmop_admin
.get("operations")
3595 # Create or append to the "_admin.operations" list
3597 "member_vnf_index": vnf_index
,
3599 "vdu_count_index": vdu_count_index
,
3600 "primitive": primitive
,
3601 "primitive_params": mapped_primitive_params
,
3604 new_op
["operationState"] = operationState
3606 new_op
["detailed-status"] = detailed_status
3608 new_op
["lcmOperationType"] = operationType
3610 new_op
["RO_nsr_id"] = RO_nsr_id
3612 new_op
["RO_scaling_info"] = RO_scaling_info
3614 # No existing operations, create key 'operations' with current operation as first list element
3615 db_nslcmop_admin
.update({"operations": [new_op
]})
3616 op_list
= db_nslcmop_admin
.get("operations")
3618 # Existing operations, append operation to list
3619 op_list
.append(new_op
)
3621 db_nslcmop_update
= {"_admin.operations": op_list
}
3622 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3623 op_index
= len(op_list
) - 1
3626 # Helper methods for scale() sub-operations
3628 # pre-scale/post-scale:
3629 # Check for 3 different cases:
3630 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3631 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3632 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3633 def _check_or_add_scale_suboperation(
3637 vnf_config_primitive
,
3641 RO_scaling_info
=None,
3643 # Find this sub-operation
3644 if RO_nsr_id
and RO_scaling_info
:
3645 operationType
= "SCALE-RO"
3647 "member_vnf_index": vnf_index
,
3648 "RO_nsr_id": RO_nsr_id
,
3649 "RO_scaling_info": RO_scaling_info
,
3653 "member_vnf_index": vnf_index
,
3654 "primitive": vnf_config_primitive
,
3655 "primitive_params": primitive_params
,
3656 "lcmOperationType": operationType
,
3658 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3659 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3660 # a. New sub-operation
3661 # The sub-operation does not exist, add it.
3662 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3663 # The following parameters are set to None for all kind of scaling:
3665 vdu_count_index
= None
3667 if RO_nsr_id
and RO_scaling_info
:
3668 vnf_config_primitive
= None
3669 primitive_params
= None
3672 RO_scaling_info
= None
3673 # Initial status for sub-operation
3674 operationState
= "PROCESSING"
3675 detailed_status
= "In progress"
3676 # Add sub-operation for pre/post-scaling (zero or more operations)
3677 self
._add
_suboperation
(
3683 vnf_config_primitive
,
3691 return self
.SUBOPERATION_STATUS_NEW
3693 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3694 # or op_index (operationState != 'COMPLETED')
3695 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3697 # Function to return execution_environment id
3699 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3700 # TODO vdu_index_count
3701 for vca
in vca_deployed_list
:
3702 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3705 async def destroy_N2VC(
3713 exec_primitives
=True,
3718 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3719 :param logging_text:
3721 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3722 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3723 :param vca_index: index in the database _admin.deployed.VCA
3724 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3725 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3726 not executed properly
3727 :param scaling_in: True destroys the application, False destroys the model
3728 :return: None or exception
3733 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3734 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3738 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3740 # execute terminate_primitives
3742 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3743 config_descriptor
.get("terminate-config-primitive"),
3744 vca_deployed
.get("ee_descriptor_id"),
3746 vdu_id
= vca_deployed
.get("vdu_id")
3747 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3748 vdu_name
= vca_deployed
.get("vdu_name")
3749 vnf_index
= vca_deployed
.get("member-vnf-index")
3750 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3751 for seq
in terminate_primitives
:
3752 # For each sequence in list, get primitive and call _ns_execute_primitive()
3753 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3754 vnf_index
, seq
.get("name")
3756 self
.logger
.debug(logging_text
+ step
)
3757 # Create the primitive for each sequence, i.e. "primitive": "touch"
3758 primitive
= seq
.get("name")
3759 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3764 self
._add
_suboperation
(
3771 mapped_primitive_params
,
3773 # Sub-operations: Call _ns_execute_primitive() instead of action()
3775 result
, result_detail
= await self
._ns
_execute
_primitive
(
3776 vca_deployed
["ee_id"],
3778 mapped_primitive_params
,
3782 except LcmException
:
3783 # this happens when VCA is not deployed. In this case it is not needed to terminate
3785 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3786 if result
not in result_ok
:
3788 "terminate_primitive {} for vnf_member_index={} fails with "
3789 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3791 # set that this VCA do not need terminated
3792 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3796 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3799 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3800 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3803 await self
.vca_map
[vca_type
].delete_execution_environment(
3804 vca_deployed
["ee_id"],
3805 scaling_in
=scaling_in
,
3810 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3811 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3812 namespace
= "." + db_nsr
["_id"]
3814 await self
.n2vc
.delete_namespace(
3815 namespace
=namespace
,
3816 total_timeout
=self
.timeout_charm_delete
,
3819 except N2VCNotFound
: # already deleted. Skip
3821 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3823 async def _terminate_RO(
3824 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
3827 Terminates a deployment from RO
3828 :param logging_text:
3829 :param nsr_deployed: db_nsr._admin.deployed
3832 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3833 this method will update only the index 2, but it will write on database the concatenated content of the list
3838 ro_nsr_id
= ro_delete_action
= None
3839 if nsr_deployed
and nsr_deployed
.get("RO"):
3840 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3841 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3844 stage
[2] = "Deleting ns from VIM."
3845 db_nsr_update
["detailed-status"] = " ".join(stage
)
3846 self
._write
_op
_status
(nslcmop_id
, stage
)
3847 self
.logger
.debug(logging_text
+ stage
[2])
3848 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3849 self
._write
_op
_status
(nslcmop_id
, stage
)
3850 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3851 ro_delete_action
= desc
["action_id"]
3853 "_admin.deployed.RO.nsr_delete_action_id"
3854 ] = ro_delete_action
3855 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3856 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3857 if ro_delete_action
:
3858 # wait until NS is deleted from VIM
3859 stage
[2] = "Waiting ns deleted from VIM."
3860 detailed_status_old
= None
3864 + " RO_id={} ro_delete_action={}".format(
3865 ro_nsr_id
, ro_delete_action
3868 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3869 self
._write
_op
_status
(nslcmop_id
, stage
)
3871 delete_timeout
= 20 * 60 # 20 minutes
3872 while delete_timeout
> 0:
3873 desc
= await self
.RO
.show(
3875 item_id_name
=ro_nsr_id
,
3876 extra_item
="action",
3877 extra_item_id
=ro_delete_action
,
3881 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3883 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3884 if ns_status
== "ERROR":
3885 raise ROclient
.ROClientException(ns_status_info
)
3886 elif ns_status
== "BUILD":
3887 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3888 elif ns_status
== "ACTIVE":
3889 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3890 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3895 ), "ROclient.check_action_status returns unknown {}".format(
3898 if stage
[2] != detailed_status_old
:
3899 detailed_status_old
= stage
[2]
3900 db_nsr_update
["detailed-status"] = " ".join(stage
)
3901 self
._write
_op
_status
(nslcmop_id
, stage
)
3902 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3903 await asyncio
.sleep(5, loop
=self
.loop
)
3905 else: # delete_timeout <= 0:
3906 raise ROclient
.ROClientException(
3907 "Timeout waiting ns deleted from VIM"
3910 except Exception as e
:
3911 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3913 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3915 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3916 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3917 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3919 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
3922 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
3924 failed_detail
.append("delete conflict: {}".format(e
))
3927 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
3930 failed_detail
.append("delete error: {}".format(e
))
3932 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
3936 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
3937 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
3939 stage
[2] = "Deleting nsd from RO."
3940 db_nsr_update
["detailed-status"] = " ".join(stage
)
3941 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3942 self
._write
_op
_status
(nslcmop_id
, stage
)
3943 await self
.RO
.delete("nsd", ro_nsd_id
)
3945 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
3947 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3948 except Exception as e
:
3950 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3952 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3954 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
3957 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
3959 failed_detail
.append(
3960 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
3962 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3964 failed_detail
.append(
3965 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
3967 self
.logger
.error(logging_text
+ failed_detail
[-1])
3969 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
3970 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
3971 if not vnf_deployed
or not vnf_deployed
["id"]:
3974 ro_vnfd_id
= vnf_deployed
["id"]
3977 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
3978 vnf_deployed
["member-vnf-index"], ro_vnfd_id
3980 db_nsr_update
["detailed-status"] = " ".join(stage
)
3981 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3982 self
._write
_op
_status
(nslcmop_id
, stage
)
3983 await self
.RO
.delete("vnfd", ro_vnfd_id
)
3985 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
3987 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3988 except Exception as e
:
3990 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3993 "_admin.deployed.RO.vnfd.{}.id".format(index
)
3997 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4000 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4002 failed_detail
.append(
4003 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4005 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4007 failed_detail
.append(
4008 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4010 self
.logger
.error(logging_text
+ failed_detail
[-1])
4013 stage
[2] = "Error deleting from VIM"
4015 stage
[2] = "Deleted from VIM"
4016 db_nsr_update
["detailed-status"] = " ".join(stage
)
4017 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4018 self
._write
_op
_status
(nslcmop_id
, stage
)
4021 raise LcmException("; ".join(failed_detail
))
4023 async def terminate(self
, nsr_id
, nslcmop_id
):
4024 # Try to lock HA task here
4025 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4026 if not task_is_locked_by_me
:
4029 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4030 self
.logger
.debug(logging_text
+ "Enter")
4031 timeout_ns_terminate
= self
.timeout_ns_terminate
4034 operation_params
= None
4036 error_list
= [] # annotates all failed error messages
4037 db_nslcmop_update
= {}
4038 autoremove
= False # autoremove after terminated
4039 tasks_dict_info
= {}
4042 "Stage 1/3: Preparing task.",
4043 "Waiting for previous operations to terminate.",
4046 # ^ contains [stage, step, VIM-status]
4048 # wait for any previous tasks in process
4049 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4051 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4052 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4053 operation_params
= db_nslcmop
.get("operationParams") or {}
4054 if operation_params
.get("timeout_ns_terminate"):
4055 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4056 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4057 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4059 db_nsr_update
["operational-status"] = "terminating"
4060 db_nsr_update
["config-status"] = "terminating"
4061 self
._write
_ns
_status
(
4063 ns_state
="TERMINATING",
4064 current_operation
="TERMINATING",
4065 current_operation_id
=nslcmop_id
,
4066 other_update
=db_nsr_update
,
4068 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4069 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4070 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4073 stage
[1] = "Getting vnf descriptors from db."
4074 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4076 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4078 db_vnfds_from_id
= {}
4079 db_vnfds_from_member_index
= {}
4081 for vnfr
in db_vnfrs_list
:
4082 vnfd_id
= vnfr
["vnfd-id"]
4083 if vnfd_id
not in db_vnfds_from_id
:
4084 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4085 db_vnfds_from_id
[vnfd_id
] = vnfd
4086 db_vnfds_from_member_index
[
4087 vnfr
["member-vnf-index-ref"]
4088 ] = db_vnfds_from_id
[vnfd_id
]
4090 # Destroy individual execution environments when there are terminating primitives.
4091 # Rest of EE will be deleted at once
4092 # TODO - check before calling _destroy_N2VC
4093 # if not operation_params.get("skip_terminate_primitives"):#
4094 # or not vca.get("needed_terminate"):
4095 stage
[0] = "Stage 2/3 execute terminating primitives."
4096 self
.logger
.debug(logging_text
+ stage
[0])
4097 stage
[1] = "Looking execution environment that needs terminate."
4098 self
.logger
.debug(logging_text
+ stage
[1])
4100 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4101 config_descriptor
= None
4102 vca_member_vnf_index
= vca
.get("member-vnf-index")
4103 vca_id
= self
.get_vca_id(
4104 db_vnfrs_dict
.get(vca_member_vnf_index
)
4105 if vca_member_vnf_index
4109 if not vca
or not vca
.get("ee_id"):
4111 if not vca
.get("member-vnf-index"):
4113 config_descriptor
= db_nsr
.get("ns-configuration")
4114 elif vca
.get("vdu_id"):
4115 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4116 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4117 elif vca
.get("kdu_name"):
4118 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4119 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4121 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4122 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4123 vca_type
= vca
.get("type")
4124 exec_terminate_primitives
= not operation_params
.get(
4125 "skip_terminate_primitives"
4126 ) and vca
.get("needed_terminate")
4127 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4128 # pending native charms
4130 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4132 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4133 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4134 task
= asyncio
.ensure_future(
4142 exec_terminate_primitives
,
4146 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4148 # wait for pending tasks of terminate primitives
4152 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4154 error_list
= await self
._wait
_for
_tasks
(
4157 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4161 tasks_dict_info
.clear()
4163 return # raise LcmException("; ".join(error_list))
4165 # remove All execution environments at once
4166 stage
[0] = "Stage 3/3 delete all."
4168 if nsr_deployed
.get("VCA"):
4169 stage
[1] = "Deleting all execution environments."
4170 self
.logger
.debug(logging_text
+ stage
[1])
4171 vca_id
= self
.get_vca_id({}, db_nsr
)
4172 task_delete_ee
= asyncio
.ensure_future(
4174 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4175 timeout
=self
.timeout_charm_delete
,
4178 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4179 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4181 # Delete from k8scluster
4182 stage
[1] = "Deleting KDUs."
4183 self
.logger
.debug(logging_text
+ stage
[1])
4184 # print(nsr_deployed)
4185 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4186 if not kdu
or not kdu
.get("kdu-instance"):
4188 kdu_instance
= kdu
.get("kdu-instance")
4189 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4190 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4191 vca_id
= self
.get_vca_id({}, db_nsr
)
4192 task_delete_kdu_instance
= asyncio
.ensure_future(
4193 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4194 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4195 kdu_instance
=kdu_instance
,
4202 + "Unknown k8s deployment type {}".format(
4203 kdu
.get("k8scluster-type")
4208 task_delete_kdu_instance
4209 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4212 stage
[1] = "Deleting ns from VIM."
4214 task_delete_ro
= asyncio
.ensure_future(
4215 self
._terminate
_ng
_ro
(
4216 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4220 task_delete_ro
= asyncio
.ensure_future(
4222 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4225 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4227 # rest of staff will be done at finally
4230 ROclient
.ROClientException
,
4235 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4237 except asyncio
.CancelledError
:
4239 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4241 exc
= "Operation was cancelled"
4242 except Exception as e
:
4243 exc
= traceback
.format_exc()
4244 self
.logger
.critical(
4245 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4250 error_list
.append(str(exc
))
4252 # wait for pending tasks
4254 stage
[1] = "Waiting for terminate pending tasks."
4255 self
.logger
.debug(logging_text
+ stage
[1])
4256 error_list
+= await self
._wait
_for
_tasks
(
4259 timeout_ns_terminate
,
4263 stage
[1] = stage
[2] = ""
4264 except asyncio
.CancelledError
:
4265 error_list
.append("Cancelled")
4266 # TODO cancell all tasks
4267 except Exception as exc
:
4268 error_list
.append(str(exc
))
4269 # update status at database
4271 error_detail
= "; ".join(error_list
)
4272 # self.logger.error(logging_text + error_detail)
4273 error_description_nslcmop
= "{} Detail: {}".format(
4274 stage
[0], error_detail
4276 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4277 nslcmop_id
, stage
[0]
4280 db_nsr_update
["operational-status"] = "failed"
4281 db_nsr_update
["detailed-status"] = (
4282 error_description_nsr
+ " Detail: " + error_detail
4284 db_nslcmop_update
["detailed-status"] = error_detail
4285 nslcmop_operation_state
= "FAILED"
4289 error_description_nsr
= error_description_nslcmop
= None
4290 ns_state
= "NOT_INSTANTIATED"
4291 db_nsr_update
["operational-status"] = "terminated"
4292 db_nsr_update
["detailed-status"] = "Done"
4293 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4294 db_nslcmop_update
["detailed-status"] = "Done"
4295 nslcmop_operation_state
= "COMPLETED"
4298 self
._write
_ns
_status
(
4301 current_operation
="IDLE",
4302 current_operation_id
=None,
4303 error_description
=error_description_nsr
,
4304 error_detail
=error_detail
,
4305 other_update
=db_nsr_update
,
4307 self
._write
_op
_status
(
4310 error_message
=error_description_nslcmop
,
4311 operation_state
=nslcmop_operation_state
,
4312 other_update
=db_nslcmop_update
,
4314 if ns_state
== "NOT_INSTANTIATED":
4318 {"nsr-id-ref": nsr_id
},
4319 {"_admin.nsState": "NOT_INSTANTIATED"},
4321 except DbException
as e
:
4324 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4328 if operation_params
:
4329 autoremove
= operation_params
.get("autoremove", False)
4330 if nslcmop_operation_state
:
4332 await self
.msg
.aiowrite(
4337 "nslcmop_id": nslcmop_id
,
4338 "operationState": nslcmop_operation_state
,
4339 "autoremove": autoremove
,
4343 except Exception as e
:
4345 logging_text
+ "kafka_write notification Exception {}".format(e
)
4348 self
.logger
.debug(logging_text
+ "Exit")
4349 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4351 async def _wait_for_tasks(
4352 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4355 error_detail_list
= []
4357 pending_tasks
= list(created_tasks_info
.keys())
4358 num_tasks
= len(pending_tasks
)
4360 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4361 self
._write
_op
_status
(nslcmop_id
, stage
)
4362 while pending_tasks
:
4364 _timeout
= timeout
+ time_start
- time()
4365 done
, pending_tasks
= await asyncio
.wait(
4366 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4368 num_done
+= len(done
)
4369 if not done
: # Timeout
4370 for task
in pending_tasks
:
4371 new_error
= created_tasks_info
[task
] + ": Timeout"
4372 error_detail_list
.append(new_error
)
4373 error_list
.append(new_error
)
4376 if task
.cancelled():
4379 exc
= task
.exception()
4381 if isinstance(exc
, asyncio
.TimeoutError
):
4383 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4384 error_list
.append(created_tasks_info
[task
])
4385 error_detail_list
.append(new_error
)
4392 ROclient
.ROClientException
,
4398 self
.logger
.error(logging_text
+ new_error
)
4400 exc_traceback
= "".join(
4401 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4405 + created_tasks_info
[task
]
4411 logging_text
+ created_tasks_info
[task
] + ": Done"
4413 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4415 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4416 if nsr_id
: # update also nsr
4421 "errorDescription": "Error at: " + ", ".join(error_list
),
4422 "errorDetail": ". ".join(error_detail_list
),
4425 self
._write
_op
_status
(nslcmop_id
, stage
)
4426 return error_detail_list
4429 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4431 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4432 The default-value is used. If it is between < > it look for a value at instantiation_params
4433 :param primitive_desc: portion of VNFD/NSD that describes primitive
4434 :param params: Params provided by user
4435 :param instantiation_params: Instantiation params provided by user
4436 :return: a dictionary with the calculated params
4438 calculated_params
= {}
4439 for parameter
in primitive_desc
.get("parameter", ()):
4440 param_name
= parameter
["name"]
4441 if param_name
in params
:
4442 calculated_params
[param_name
] = params
[param_name
]
4443 elif "default-value" in parameter
or "value" in parameter
:
4444 if "value" in parameter
:
4445 calculated_params
[param_name
] = parameter
["value"]
4447 calculated_params
[param_name
] = parameter
["default-value"]
4449 isinstance(calculated_params
[param_name
], str)
4450 and calculated_params
[param_name
].startswith("<")
4451 and calculated_params
[param_name
].endswith(">")
4453 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4454 calculated_params
[param_name
] = instantiation_params
[
4455 calculated_params
[param_name
][1:-1]
4459 "Parameter {} needed to execute primitive {} not provided".format(
4460 calculated_params
[param_name
], primitive_desc
["name"]
4465 "Parameter {} needed to execute primitive {} not provided".format(
4466 param_name
, primitive_desc
["name"]
4470 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4471 calculated_params
[param_name
] = yaml
.safe_dump(
4472 calculated_params
[param_name
], default_flow_style
=True, width
=256
4474 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4476 ].startswith("!!yaml "):
4477 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4478 if parameter
.get("data-type") == "INTEGER":
4480 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4481 except ValueError: # error converting string to int
4483 "Parameter {} of primitive {} must be integer".format(
4484 param_name
, primitive_desc
["name"]
4487 elif parameter
.get("data-type") == "BOOLEAN":
4488 calculated_params
[param_name
] = not (
4489 (str(calculated_params
[param_name
])).lower() == "false"
4492 # add always ns_config_info if primitive name is config
4493 if primitive_desc
["name"] == "config":
4494 if "ns_config_info" in instantiation_params
:
4495 calculated_params
["ns_config_info"] = instantiation_params
[
4498 return calculated_params
4500 def _look_for_deployed_vca(
4507 ee_descriptor_id
=None,
4509 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4510 for vca
in deployed_vca
:
4513 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4516 vdu_count_index
is not None
4517 and vdu_count_index
!= vca
["vdu_count_index"]
4520 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4522 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4526 # vca_deployed not found
4528 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4529 " is not deployed".format(
4538 ee_id
= vca
.get("ee_id")
4540 "type", "lxc_proxy_charm"
4541 ) # default value for backward compatibility - proxy charm
4544 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4545 "execution environment".format(
4546 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4549 return ee_id
, vca_type
4551 async def _ns_execute_primitive(
4557 retries_interval
=30,
4564 if primitive
== "config":
4565 primitive_params
= {"params": primitive_params
}
4567 vca_type
= vca_type
or "lxc_proxy_charm"
4571 output
= await asyncio
.wait_for(
4572 self
.vca_map
[vca_type
].exec_primitive(
4574 primitive_name
=primitive
,
4575 params_dict
=primitive_params
,
4576 progress_timeout
=self
.timeout_progress_primitive
,
4577 total_timeout
=self
.timeout_primitive
,
4582 timeout
=timeout
or self
.timeout_primitive
,
4586 except asyncio
.CancelledError
:
4588 except Exception as e
: # asyncio.TimeoutError
4589 if isinstance(e
, asyncio
.TimeoutError
):
4594 "Error executing action {} on {} -> {}".format(
4599 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4601 return "FAILED", str(e
)
4603 return "COMPLETED", output
4605 except (LcmException
, asyncio
.CancelledError
):
4607 except Exception as e
:
4608 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4610 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4612 Updating the vca_status with latest juju information in nsrs record
4613 :param: nsr_id: Id of the nsr
4614 :param: nslcmop_id: Id of the nslcmop
4618 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4619 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4620 vca_id
= self
.get_vca_id({}, db_nsr
)
4621 if db_nsr
["_admin"]["deployed"]["K8s"]:
4622 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4623 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4624 await self
._on
_update
_k
8s
_db
(
4625 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4628 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4629 table
, filter = "nsrs", {"_id": nsr_id
}
4630 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4631 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4633 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4634 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4636 async def action(self
, nsr_id
, nslcmop_id
):
4637 # Try to lock HA task here
4638 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4639 if not task_is_locked_by_me
:
4642 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4643 self
.logger
.debug(logging_text
+ "Enter")
4644 # get all needed from database
4648 db_nslcmop_update
= {}
4649 nslcmop_operation_state
= None
4650 error_description_nslcmop
= None
4653 # wait for any previous tasks in process
4654 step
= "Waiting for previous operations to terminate"
4655 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4657 self
._write
_ns
_status
(
4660 current_operation
="RUNNING ACTION",
4661 current_operation_id
=nslcmop_id
,
4664 step
= "Getting information from database"
4665 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4666 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4667 if db_nslcmop
["operationParams"].get("primitive_params"):
4668 db_nslcmop
["operationParams"]["primitive_params"] = json
.loads(
4669 db_nslcmop
["operationParams"]["primitive_params"]
4672 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4673 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4674 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4675 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4676 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4677 primitive
= db_nslcmop
["operationParams"]["primitive"]
4678 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4679 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4680 "timeout_ns_action", self
.timeout_primitive
4684 step
= "Getting vnfr from database"
4685 db_vnfr
= self
.db
.get_one(
4686 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4688 if db_vnfr
.get("kdur"):
4690 for kdur
in db_vnfr
["kdur"]:
4691 if kdur
.get("additionalParams"):
4692 kdur
["additionalParams"] = json
.loads(
4693 kdur
["additionalParams"]
4695 kdur_list
.append(kdur
)
4696 db_vnfr
["kdur"] = kdur_list
4697 step
= "Getting vnfd from database"
4698 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4700 step
= "Getting nsd from database"
4701 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4703 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4704 # for backward compatibility
4705 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4706 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4707 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4708 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4710 # look for primitive
4711 config_primitive_desc
= descriptor_configuration
= None
4713 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4715 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4717 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4719 descriptor_configuration
= db_nsd
.get("ns-configuration")
4721 if descriptor_configuration
and descriptor_configuration
.get(
4724 for config_primitive
in descriptor_configuration
["config-primitive"]:
4725 if config_primitive
["name"] == primitive
:
4726 config_primitive_desc
= config_primitive
4729 if not config_primitive_desc
:
4730 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4732 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4736 primitive_name
= primitive
4737 ee_descriptor_id
= None
4739 primitive_name
= config_primitive_desc
.get(
4740 "execution-environment-primitive", primitive
4742 ee_descriptor_id
= config_primitive_desc
.get(
4743 "execution-environment-ref"
4749 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4751 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4754 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4756 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4758 desc_params
= parse_yaml_strings(
4759 db_vnfr
.get("additionalParamsForVnf")
4762 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4763 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4764 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4766 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4767 actions
.add(primitive
["name"])
4768 for primitive
in kdu_configuration
.get("config-primitive", []):
4769 actions
.add(primitive
["name"])
4770 kdu_action
= True if primitive_name
in actions
else False
4772 # TODO check if ns is in a proper status
4774 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4776 # kdur and desc_params already set from before
4777 if primitive_params
:
4778 desc_params
.update(primitive_params
)
4779 # TODO Check if we will need something at vnf level
4780 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4782 kdu_name
== kdu
["kdu-name"]
4783 and kdu
["member-vnf-index"] == vnf_index
4788 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4791 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4792 msg
= "unknown k8scluster-type '{}'".format(
4793 kdu
.get("k8scluster-type")
4795 raise LcmException(msg
)
4798 "collection": "nsrs",
4799 "filter": {"_id": nsr_id
},
4800 "path": "_admin.deployed.K8s.{}".format(index
),
4804 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4806 step
= "Executing kdu {}".format(primitive_name
)
4807 if primitive_name
== "upgrade":
4808 if desc_params
.get("kdu_model"):
4809 kdu_model
= desc_params
.get("kdu_model")
4810 del desc_params
["kdu_model"]
4812 kdu_model
= kdu
.get("kdu-model")
4813 parts
= kdu_model
.split(sep
=":")
4815 kdu_model
= parts
[0]
4817 detailed_status
= await asyncio
.wait_for(
4818 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4819 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4820 kdu_instance
=kdu
.get("kdu-instance"),
4822 kdu_model
=kdu_model
,
4825 timeout
=timeout_ns_action
,
4827 timeout
=timeout_ns_action
+ 10,
4830 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4832 elif primitive_name
== "rollback":
4833 detailed_status
= await asyncio
.wait_for(
4834 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4835 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4836 kdu_instance
=kdu
.get("kdu-instance"),
4839 timeout
=timeout_ns_action
,
4841 elif primitive_name
== "status":
4842 detailed_status
= await asyncio
.wait_for(
4843 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4844 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4845 kdu_instance
=kdu
.get("kdu-instance"),
4848 timeout
=timeout_ns_action
,
4851 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
4852 kdu
["kdu-name"], nsr_id
4854 params
= self
._map
_primitive
_params
(
4855 config_primitive_desc
, primitive_params
, desc_params
4858 detailed_status
= await asyncio
.wait_for(
4859 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
4860 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4861 kdu_instance
=kdu_instance
,
4862 primitive_name
=primitive_name
,
4865 timeout
=timeout_ns_action
,
4868 timeout
=timeout_ns_action
,
4872 nslcmop_operation_state
= "COMPLETED"
4874 detailed_status
= ""
4875 nslcmop_operation_state
= "FAILED"
4877 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
4878 nsr_deployed
["VCA"],
4879 member_vnf_index
=vnf_index
,
4881 vdu_count_index
=vdu_count_index
,
4882 ee_descriptor_id
=ee_descriptor_id
,
4884 for vca_index
, vca_deployed
in enumerate(
4885 db_nsr
["_admin"]["deployed"]["VCA"]
4887 if vca_deployed
.get("member-vnf-index") == vnf_index
:
4889 "collection": "nsrs",
4890 "filter": {"_id": nsr_id
},
4891 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
4895 nslcmop_operation_state
,
4897 ) = await self
._ns
_execute
_primitive
(
4899 primitive
=primitive_name
,
4900 primitive_params
=self
._map
_primitive
_params
(
4901 config_primitive_desc
, primitive_params
, desc_params
4903 timeout
=timeout_ns_action
,
4909 db_nslcmop_update
["detailed-status"] = detailed_status
4910 error_description_nslcmop
= (
4911 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
4915 + " task Done with result {} {}".format(
4916 nslcmop_operation_state
, detailed_status
4919 return # database update is called inside finally
4921 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
4922 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4924 except asyncio
.CancelledError
:
4926 logging_text
+ "Cancelled Exception while '{}'".format(step
)
4928 exc
= "Operation was cancelled"
4929 except asyncio
.TimeoutError
:
4930 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
4932 except Exception as e
:
4933 exc
= traceback
.format_exc()
4934 self
.logger
.critical(
4935 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
4944 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
4945 nslcmop_operation_state
= "FAILED"
4947 self
._write
_ns
_status
(
4951 ], # TODO check if degraded. For the moment use previous status
4952 current_operation
="IDLE",
4953 current_operation_id
=None,
4954 # error_description=error_description_nsr,
4955 # error_detail=error_detail,
4956 other_update
=db_nsr_update
,
4959 self
._write
_op
_status
(
4962 error_message
=error_description_nslcmop
,
4963 operation_state
=nslcmop_operation_state
,
4964 other_update
=db_nslcmop_update
,
4967 if nslcmop_operation_state
:
4969 await self
.msg
.aiowrite(
4974 "nslcmop_id": nslcmop_id
,
4975 "operationState": nslcmop_operation_state
,
4979 except Exception as e
:
4981 logging_text
+ "kafka_write notification Exception {}".format(e
)
4983 self
.logger
.debug(logging_text
+ "Exit")
4984 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
4985 return nslcmop_operation_state
, detailed_status
4987 async def scale(self
, nsr_id
, nslcmop_id
):
4988 # Try to lock HA task here
4989 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4990 if not task_is_locked_by_me
:
4993 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
4994 stage
= ["", "", ""]
4995 tasks_dict_info
= {}
4996 # ^ stage, step, VIM progress
4997 self
.logger
.debug(logging_text
+ "Enter")
4998 # get all needed from database
5000 db_nslcmop_update
= {}
5003 # in case of error, indicates what part of scale was failed to put nsr at error status
5004 scale_process
= None
5005 old_operational_status
= ""
5006 old_config_status
= ""
5009 # wait for any previous tasks in process
5010 step
= "Waiting for previous operations to terminate"
5011 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5012 self
._write
_ns
_status
(
5015 current_operation
="SCALING",
5016 current_operation_id
=nslcmop_id
,
5019 step
= "Getting nslcmop from database"
5021 step
+ " after having waited for previous tasks to be completed"
5023 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5025 step
= "Getting nsr from database"
5026 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5027 old_operational_status
= db_nsr
["operational-status"]
5028 old_config_status
= db_nsr
["config-status"]
5030 step
= "Parsing scaling parameters"
5031 db_nsr_update
["operational-status"] = "scaling"
5032 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5033 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5035 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5037 ]["member-vnf-index"]
5038 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5040 ]["scaling-group-descriptor"]
5041 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5042 # for backward compatibility
5043 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5044 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5045 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5046 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5048 step
= "Getting vnfr from database"
5049 db_vnfr
= self
.db
.get_one(
5050 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5053 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5055 step
= "Getting vnfd from database"
5056 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5058 base_folder
= db_vnfd
["_admin"]["storage"]
5060 step
= "Getting scaling-group-descriptor"
5061 scaling_descriptor
= find_in_list(
5062 get_scaling_aspect(db_vnfd
),
5063 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5065 if not scaling_descriptor
:
5067 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5068 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5071 step
= "Sending scale order to VIM"
5072 # TODO check if ns is in a proper status
5074 if not db_nsr
["_admin"].get("scaling-group"):
5079 "_admin.scaling-group": [
5080 {"name": scaling_group
, "nb-scale-op": 0}
5084 admin_scale_index
= 0
5086 for admin_scale_index
, admin_scale_info
in enumerate(
5087 db_nsr
["_admin"]["scaling-group"]
5089 if admin_scale_info
["name"] == scaling_group
:
5090 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5092 else: # not found, set index one plus last element and add new entry with the name
5093 admin_scale_index
+= 1
5095 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5098 vca_scaling_info
= []
5099 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5100 if scaling_type
== "SCALE_OUT":
5101 if "aspect-delta-details" not in scaling_descriptor
:
5103 "Aspect delta details not fount in scaling descriptor {}".format(
5104 scaling_descriptor
["name"]
5107 # count if max-instance-count is reached
5108 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5110 scaling_info
["scaling_direction"] = "OUT"
5111 scaling_info
["vdu-create"] = {}
5112 scaling_info
["kdu-create"] = {}
5113 for delta
in deltas
:
5114 for vdu_delta
in delta
.get("vdu-delta", {}):
5115 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5116 # vdu_index also provides the number of instance of the targeted vdu
5117 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5118 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5122 additional_params
= (
5123 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5126 cloud_init_list
= []
5128 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5129 max_instance_count
= 10
5130 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5131 max_instance_count
= vdu_profile
.get(
5132 "max-number-of-instances", 10
5135 default_instance_num
= get_number_of_instances(
5138 instances_number
= vdu_delta
.get("number-of-instances", 1)
5139 nb_scale_op
+= instances_number
5141 new_instance_count
= nb_scale_op
+ default_instance_num
5142 # Control if new count is over max and vdu count is less than max.
5143 # Then assign new instance count
5144 if new_instance_count
> max_instance_count
> vdu_count
:
5145 instances_number
= new_instance_count
- max_instance_count
5147 instances_number
= instances_number
5149 if new_instance_count
> max_instance_count
:
5151 "reached the limit of {} (max-instance-count) "
5152 "scaling-out operations for the "
5153 "scaling-group-descriptor '{}'".format(
5154 nb_scale_op
, scaling_group
5157 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5159 # TODO Information of its own ip is not available because db_vnfr is not updated.
5160 additional_params
["OSM"] = get_osm_params(
5161 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5163 cloud_init_list
.append(
5164 self
._parse
_cloud
_init
(
5171 vca_scaling_info
.append(
5173 "osm_vdu_id": vdu_delta
["id"],
5174 "member-vnf-index": vnf_index
,
5176 "vdu_index": vdu_index
+ x
,
5179 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5180 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5181 kdu_profile
= get_kdu_profile(db_vnfd
, kdu_delta
["id"])
5182 kdu_name
= kdu_profile
["kdu-name"]
5183 resource_name
= kdu_profile
["resource-name"]
5185 # Might have different kdus in the same delta
5186 # Should have list for each kdu
5187 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5188 scaling_info
["kdu-create"][kdu_name
] = []
5190 kdur
= get_kdur(db_vnfr
, kdu_name
)
5191 if kdur
.get("helm-chart"):
5192 k8s_cluster_type
= "helm-chart-v3"
5193 self
.logger
.debug("kdur: {}".format(kdur
))
5195 kdur
.get("helm-version")
5196 and kdur
.get("helm-version") == "v2"
5198 k8s_cluster_type
= "helm-chart"
5199 raise NotImplementedError
5200 elif kdur
.get("juju-bundle"):
5201 k8s_cluster_type
= "juju-bundle"
5204 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5205 "juju-bundle. Maybe an old NBI version is running".format(
5206 db_vnfr
["member-vnf-index-ref"], kdu_name
5210 max_instance_count
= 10
5211 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5212 max_instance_count
= kdu_profile
.get(
5213 "max-number-of-instances", 10
5216 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5217 deployed_kdu
, _
= get_deployed_kdu(
5218 nsr_deployed
, kdu_name
, vnf_index
5220 if deployed_kdu
is None:
5222 "KDU '{}' for vnf '{}' not deployed".format(
5226 kdu_instance
= deployed_kdu
.get("kdu-instance")
5227 instance_num
= await self
.k8scluster_map
[
5229 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5230 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5231 "number-of-instances", 1
5234 # Control if new count is over max and instance_num is less than max.
5235 # Then assign max instance number to kdu replica count
5236 if kdu_replica_count
> max_instance_count
> instance_num
:
5237 kdu_replica_count
= max_instance_count
5238 if kdu_replica_count
> max_instance_count
:
5240 "reached the limit of {} (max-instance-count) "
5241 "scaling-out operations for the "
5242 "scaling-group-descriptor '{}'".format(
5243 instance_num
, scaling_group
5247 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5248 vca_scaling_info
.append(
5250 "osm_kdu_id": kdu_name
,
5251 "member-vnf-index": vnf_index
,
5253 "kdu_index": instance_num
+ x
- 1,
5256 scaling_info
["kdu-create"][kdu_name
].append(
5258 "member-vnf-index": vnf_index
,
5260 "k8s-cluster-type": k8s_cluster_type
,
5261 "resource-name": resource_name
,
5262 "scale": kdu_replica_count
,
5265 elif scaling_type
== "SCALE_IN":
5266 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5268 scaling_info
["scaling_direction"] = "IN"
5269 scaling_info
["vdu-delete"] = {}
5270 scaling_info
["kdu-delete"] = {}
5272 for delta
in deltas
:
5273 for vdu_delta
in delta
.get("vdu-delta", {}):
5274 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5275 min_instance_count
= 0
5276 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5277 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5278 min_instance_count
= vdu_profile
["min-number-of-instances"]
5280 default_instance_num
= get_number_of_instances(
5281 db_vnfd
, vdu_delta
["id"]
5283 instance_num
= vdu_delta
.get("number-of-instances", 1)
5284 nb_scale_op
-= instance_num
5286 new_instance_count
= nb_scale_op
+ default_instance_num
5288 if new_instance_count
< min_instance_count
< vdu_count
:
5289 instances_number
= min_instance_count
- new_instance_count
5291 instances_number
= instance_num
5293 if new_instance_count
< min_instance_count
:
5295 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5296 "scaling-group-descriptor '{}'".format(
5297 nb_scale_op
, scaling_group
5300 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5301 vca_scaling_info
.append(
5303 "osm_vdu_id": vdu_delta
["id"],
5304 "member-vnf-index": vnf_index
,
5306 "vdu_index": vdu_index
- 1 - x
,
5309 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5310 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5311 kdu_profile
= get_kdu_profile(db_vnfd
, kdu_delta
["id"])
5312 kdu_name
= kdu_profile
["kdu-name"]
5313 resource_name
= kdu_profile
["resource-name"]
5315 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5316 scaling_info
["kdu-delete"][kdu_name
] = []
5318 kdur
= get_kdur(db_vnfr
, kdu_name
)
5319 if kdur
.get("helm-chart"):
5320 k8s_cluster_type
= "helm-chart-v3"
5321 self
.logger
.debug("kdur: {}".format(kdur
))
5323 kdur
.get("helm-version")
5324 and kdur
.get("helm-version") == "v2"
5326 k8s_cluster_type
= "helm-chart"
5327 raise NotImplementedError
5328 elif kdur
.get("juju-bundle"):
5329 k8s_cluster_type
= "juju-bundle"
5332 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5333 "juju-bundle. Maybe an old NBI version is running".format(
5334 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5338 min_instance_count
= 0
5339 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5340 min_instance_count
= kdu_profile
["min-number-of-instances"]
5342 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5343 deployed_kdu
, _
= get_deployed_kdu(
5344 nsr_deployed
, kdu_name
, vnf_index
5346 if deployed_kdu
is None:
5348 "KDU '{}' for vnf '{}' not deployed".format(
5352 kdu_instance
= deployed_kdu
.get("kdu-instance")
5353 instance_num
= await self
.k8scluster_map
[
5355 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5356 kdu_replica_count
= instance_num
- kdu_delta
.get(
5357 "number-of-instances", 1
5360 if kdu_replica_count
< min_instance_count
< instance_num
:
5361 kdu_replica_count
= min_instance_count
5362 if kdu_replica_count
< min_instance_count
:
5364 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5365 "scaling-group-descriptor '{}'".format(
5366 instance_num
, scaling_group
5370 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5371 vca_scaling_info
.append(
5373 "osm_kdu_id": kdu_name
,
5374 "member-vnf-index": vnf_index
,
5376 "kdu_index": instance_num
- x
- 1,
5379 scaling_info
["kdu-delete"][kdu_name
].append(
5381 "member-vnf-index": vnf_index
,
5383 "k8s-cluster-type": k8s_cluster_type
,
5384 "resource-name": resource_name
,
5385 "scale": kdu_replica_count
,
5389 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5390 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5391 if scaling_info
["scaling_direction"] == "IN":
5392 for vdur
in reversed(db_vnfr
["vdur"]):
5393 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5394 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5395 scaling_info
["vdu"].append(
5397 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5398 "vdu_id": vdur
["vdu-id-ref"],
5402 for interface
in vdur
["interfaces"]:
5403 scaling_info
["vdu"][-1]["interface"].append(
5405 "name": interface
["name"],
5406 "ip_address": interface
["ip-address"],
5407 "mac_address": interface
.get("mac-address"),
5410 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5413 step
= "Executing pre-scale vnf-config-primitive"
5414 if scaling_descriptor
.get("scaling-config-action"):
5415 for scaling_config_action
in scaling_descriptor
[
5416 "scaling-config-action"
5419 scaling_config_action
.get("trigger") == "pre-scale-in"
5420 and scaling_type
== "SCALE_IN"
5422 scaling_config_action
.get("trigger") == "pre-scale-out"
5423 and scaling_type
== "SCALE_OUT"
5425 vnf_config_primitive
= scaling_config_action
[
5426 "vnf-config-primitive-name-ref"
5428 step
= db_nslcmop_update
[
5430 ] = "executing pre-scale scaling-config-action '{}'".format(
5431 vnf_config_primitive
5434 # look for primitive
5435 for config_primitive
in (
5436 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5437 ).get("config-primitive", ()):
5438 if config_primitive
["name"] == vnf_config_primitive
:
5442 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5443 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5444 "primitive".format(scaling_group
, vnf_config_primitive
)
5447 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5448 if db_vnfr
.get("additionalParamsForVnf"):
5449 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5451 scale_process
= "VCA"
5452 db_nsr_update
["config-status"] = "configuring pre-scaling"
5453 primitive_params
= self
._map
_primitive
_params
(
5454 config_primitive
, {}, vnfr_params
5457 # Pre-scale retry check: Check if this sub-operation has been executed before
5458 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5461 vnf_config_primitive
,
5465 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5466 # Skip sub-operation
5467 result
= "COMPLETED"
5468 result_detail
= "Done"
5471 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5472 vnf_config_primitive
, result
, result_detail
5476 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5477 # New sub-operation: Get index of this sub-operation
5479 len(db_nslcmop
.get("_admin", {}).get("operations"))
5484 + "vnf_config_primitive={} New sub-operation".format(
5485 vnf_config_primitive
5489 # retry: Get registered params for this existing sub-operation
5490 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5493 vnf_index
= op
.get("member_vnf_index")
5494 vnf_config_primitive
= op
.get("primitive")
5495 primitive_params
= op
.get("primitive_params")
5498 + "vnf_config_primitive={} Sub-operation retry".format(
5499 vnf_config_primitive
5502 # Execute the primitive, either with new (first-time) or registered (reintent) args
5503 ee_descriptor_id
= config_primitive
.get(
5504 "execution-environment-ref"
5506 primitive_name
= config_primitive
.get(
5507 "execution-environment-primitive", vnf_config_primitive
5509 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5510 nsr_deployed
["VCA"],
5511 member_vnf_index
=vnf_index
,
5513 vdu_count_index
=None,
5514 ee_descriptor_id
=ee_descriptor_id
,
5516 result
, result_detail
= await self
._ns
_execute
_primitive
(
5525 + "vnf_config_primitive={} Done with result {} {}".format(
5526 vnf_config_primitive
, result
, result_detail
5529 # Update operationState = COMPLETED | FAILED
5530 self
._update
_suboperation
_status
(
5531 db_nslcmop
, op_index
, result
, result_detail
5534 if result
== "FAILED":
5535 raise LcmException(result_detail
)
5536 db_nsr_update
["config-status"] = old_config_status
5537 scale_process
= None
5541 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5544 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5547 # SCALE-IN VCA - BEGIN
5548 if vca_scaling_info
:
5549 step
= db_nslcmop_update
[
5551 ] = "Deleting the execution environments"
5552 scale_process
= "VCA"
5553 for vca_info
in vca_scaling_info
:
5554 if vca_info
["type"] == "delete":
5555 member_vnf_index
= str(vca_info
["member-vnf-index"])
5557 logging_text
+ "vdu info: {}".format(vca_info
)
5559 if vca_info
.get("osm_vdu_id"):
5560 vdu_id
= vca_info
["osm_vdu_id"]
5561 vdu_index
= int(vca_info
["vdu_index"])
5564 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5565 member_vnf_index
, vdu_id
, vdu_index
5569 kdu_id
= vca_info
["osm_kdu_id"]
5572 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5573 member_vnf_index
, kdu_id
, vdu_index
5575 stage
[2] = step
= "Scaling in VCA"
5576 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5577 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5578 config_update
= db_nsr
["configurationStatus"]
5579 for vca_index
, vca
in enumerate(vca_update
):
5581 (vca
or vca
.get("ee_id"))
5582 and vca
["member-vnf-index"] == member_vnf_index
5583 and vca
["vdu_count_index"] == vdu_index
5585 if vca
.get("vdu_id"):
5586 config_descriptor
= get_configuration(
5587 db_vnfd
, vca
.get("vdu_id")
5589 elif vca
.get("kdu_name"):
5590 config_descriptor
= get_configuration(
5591 db_vnfd
, vca
.get("kdu_name")
5594 config_descriptor
= get_configuration(
5595 db_vnfd
, db_vnfd
["id"]
5597 operation_params
= (
5598 db_nslcmop
.get("operationParams") or {}
5600 exec_terminate_primitives
= not operation_params
.get(
5601 "skip_terminate_primitives"
5602 ) and vca
.get("needed_terminate")
5603 task
= asyncio
.ensure_future(
5612 exec_primitives
=exec_terminate_primitives
,
5616 timeout
=self
.timeout_charm_delete
,
5619 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5622 del vca_update
[vca_index
]
5623 del config_update
[vca_index
]
5624 # wait for pending tasks of terminate primitives
5628 + "Waiting for tasks {}".format(
5629 list(tasks_dict_info
.keys())
5632 error_list
= await self
._wait
_for
_tasks
(
5636 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5641 tasks_dict_info
.clear()
5643 raise LcmException("; ".join(error_list
))
5645 db_vca_and_config_update
= {
5646 "_admin.deployed.VCA": vca_update
,
5647 "configurationStatus": config_update
,
5650 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5652 scale_process
= None
5653 # SCALE-IN VCA - END
5656 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5657 scale_process
= "RO"
5658 if self
.ro_config
.get("ng"):
5659 await self
._scale
_ng
_ro
(
5660 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5662 scaling_info
.pop("vdu-create", None)
5663 scaling_info
.pop("vdu-delete", None)
5665 scale_process
= None
5669 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5670 scale_process
= "KDU"
5671 await self
._scale
_kdu
(
5672 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5674 scaling_info
.pop("kdu-create", None)
5675 scaling_info
.pop("kdu-delete", None)
5677 scale_process
= None
5681 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5683 # SCALE-UP VCA - BEGIN
5684 if vca_scaling_info
:
5685 step
= db_nslcmop_update
[
5687 ] = "Creating new execution environments"
5688 scale_process
= "VCA"
5689 for vca_info
in vca_scaling_info
:
5690 if vca_info
["type"] == "create":
5691 member_vnf_index
= str(vca_info
["member-vnf-index"])
5693 logging_text
+ "vdu info: {}".format(vca_info
)
5695 vnfd_id
= db_vnfr
["vnfd-ref"]
5696 if vca_info
.get("osm_vdu_id"):
5697 vdu_index
= int(vca_info
["vdu_index"])
5698 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5699 if db_vnfr
.get("additionalParamsForVnf"):
5700 deploy_params
.update(
5702 db_vnfr
["additionalParamsForVnf"].copy()
5705 descriptor_config
= get_configuration(
5706 db_vnfd
, db_vnfd
["id"]
5708 if descriptor_config
:
5713 logging_text
=logging_text
5714 + "member_vnf_index={} ".format(member_vnf_index
),
5717 nslcmop_id
=nslcmop_id
,
5723 member_vnf_index
=member_vnf_index
,
5724 vdu_index
=vdu_index
,
5726 deploy_params
=deploy_params
,
5727 descriptor_config
=descriptor_config
,
5728 base_folder
=base_folder
,
5729 task_instantiation_info
=tasks_dict_info
,
5732 vdu_id
= vca_info
["osm_vdu_id"]
5733 vdur
= find_in_list(
5734 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5736 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5737 if vdur
.get("additionalParams"):
5738 deploy_params_vdu
= parse_yaml_strings(
5739 vdur
["additionalParams"]
5742 deploy_params_vdu
= deploy_params
5743 deploy_params_vdu
["OSM"] = get_osm_params(
5744 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5746 if descriptor_config
:
5751 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5752 member_vnf_index
, vdu_id
, vdu_index
5754 stage
[2] = step
= "Scaling out VCA"
5755 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5757 logging_text
=logging_text
5758 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5759 member_vnf_index
, vdu_id
, vdu_index
5763 nslcmop_id
=nslcmop_id
,
5769 member_vnf_index
=member_vnf_index
,
5770 vdu_index
=vdu_index
,
5772 deploy_params
=deploy_params_vdu
,
5773 descriptor_config
=descriptor_config
,
5774 base_folder
=base_folder
,
5775 task_instantiation_info
=tasks_dict_info
,
5779 kdu_name
= vca_info
["osm_kdu_id"]
5780 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5781 if descriptor_config
:
5783 kdu_index
= int(vca_info
["kdu_index"])
5787 for x
in db_vnfr
["kdur"]
5788 if x
["kdu-name"] == kdu_name
5790 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5791 if kdur
.get("additionalParams"):
5792 deploy_params_kdu
= parse_yaml_strings(
5793 kdur
["additionalParams"]
5797 logging_text
=logging_text
,
5800 nslcmop_id
=nslcmop_id
,
5806 member_vnf_index
=member_vnf_index
,
5807 vdu_index
=kdu_index
,
5809 deploy_params
=deploy_params_kdu
,
5810 descriptor_config
=descriptor_config
,
5811 base_folder
=base_folder
,
5812 task_instantiation_info
=tasks_dict_info
,
5815 # SCALE-UP VCA - END
5816 scale_process
= None
5819 # execute primitive service POST-SCALING
5820 step
= "Executing post-scale vnf-config-primitive"
5821 if scaling_descriptor
.get("scaling-config-action"):
5822 for scaling_config_action
in scaling_descriptor
[
5823 "scaling-config-action"
5826 scaling_config_action
.get("trigger") == "post-scale-in"
5827 and scaling_type
== "SCALE_IN"
5829 scaling_config_action
.get("trigger") == "post-scale-out"
5830 and scaling_type
== "SCALE_OUT"
5832 vnf_config_primitive
= scaling_config_action
[
5833 "vnf-config-primitive-name-ref"
5835 step
= db_nslcmop_update
[
5837 ] = "executing post-scale scaling-config-action '{}'".format(
5838 vnf_config_primitive
5841 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5842 if db_vnfr
.get("additionalParamsForVnf"):
5843 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5845 # look for primitive
5846 for config_primitive
in (
5847 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5848 ).get("config-primitive", ()):
5849 if config_primitive
["name"] == vnf_config_primitive
:
5853 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
5854 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
5855 "config-primitive".format(
5856 scaling_group
, vnf_config_primitive
5859 scale_process
= "VCA"
5860 db_nsr_update
["config-status"] = "configuring post-scaling"
5861 primitive_params
= self
._map
_primitive
_params
(
5862 config_primitive
, {}, vnfr_params
5865 # Post-scale retry check: Check if this sub-operation has been executed before
5866 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5869 vnf_config_primitive
,
5873 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5874 # Skip sub-operation
5875 result
= "COMPLETED"
5876 result_detail
= "Done"
5879 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5880 vnf_config_primitive
, result
, result_detail
5884 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5885 # New sub-operation: Get index of this sub-operation
5887 len(db_nslcmop
.get("_admin", {}).get("operations"))
5892 + "vnf_config_primitive={} New sub-operation".format(
5893 vnf_config_primitive
5897 # retry: Get registered params for this existing sub-operation
5898 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5901 vnf_index
= op
.get("member_vnf_index")
5902 vnf_config_primitive
= op
.get("primitive")
5903 primitive_params
= op
.get("primitive_params")
5906 + "vnf_config_primitive={} Sub-operation retry".format(
5907 vnf_config_primitive
5910 # Execute the primitive, either with new (first-time) or registered (reintent) args
5911 ee_descriptor_id
= config_primitive
.get(
5912 "execution-environment-ref"
5914 primitive_name
= config_primitive
.get(
5915 "execution-environment-primitive", vnf_config_primitive
5917 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5918 nsr_deployed
["VCA"],
5919 member_vnf_index
=vnf_index
,
5921 vdu_count_index
=None,
5922 ee_descriptor_id
=ee_descriptor_id
,
5924 result
, result_detail
= await self
._ns
_execute
_primitive
(
5933 + "vnf_config_primitive={} Done with result {} {}".format(
5934 vnf_config_primitive
, result
, result_detail
5937 # Update operationState = COMPLETED | FAILED
5938 self
._update
_suboperation
_status
(
5939 db_nslcmop
, op_index
, result
, result_detail
5942 if result
== "FAILED":
5943 raise LcmException(result_detail
)
5944 db_nsr_update
["config-status"] = old_config_status
5945 scale_process
= None
5950 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
5951 db_nsr_update
["operational-status"] = (
5953 if old_operational_status
== "failed"
5954 else old_operational_status
5956 db_nsr_update
["config-status"] = old_config_status
5959 ROclient
.ROClientException
,
5964 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5966 except asyncio
.CancelledError
:
5968 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5970 exc
= "Operation was cancelled"
5971 except Exception as e
:
5972 exc
= traceback
.format_exc()
5973 self
.logger
.critical(
5974 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5978 self
._write
_ns
_status
(
5981 current_operation
="IDLE",
5982 current_operation_id
=None,
5985 stage
[1] = "Waiting for instantiate pending tasks."
5986 self
.logger
.debug(logging_text
+ stage
[1])
5987 exc
= await self
._wait
_for
_tasks
(
5990 self
.timeout_ns_deploy
,
5998 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5999 nslcmop_operation_state
= "FAILED"
6001 db_nsr_update
["operational-status"] = old_operational_status
6002 db_nsr_update
["config-status"] = old_config_status
6003 db_nsr_update
["detailed-status"] = ""
6005 if "VCA" in scale_process
:
6006 db_nsr_update
["config-status"] = "failed"
6007 if "RO" in scale_process
:
6008 db_nsr_update
["operational-status"] = "failed"
6011 ] = "FAILED scaling nslcmop={} {}: {}".format(
6012 nslcmop_id
, step
, exc
6015 error_description_nslcmop
= None
6016 nslcmop_operation_state
= "COMPLETED"
6017 db_nslcmop_update
["detailed-status"] = "Done"
6019 self
._write
_op
_status
(
6022 error_message
=error_description_nslcmop
,
6023 operation_state
=nslcmop_operation_state
,
6024 other_update
=db_nslcmop_update
,
6027 self
._write
_ns
_status
(
6030 current_operation
="IDLE",
6031 current_operation_id
=None,
6032 other_update
=db_nsr_update
,
6035 if nslcmop_operation_state
:
6039 "nslcmop_id": nslcmop_id
,
6040 "operationState": nslcmop_operation_state
,
6042 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6043 except Exception as e
:
6045 logging_text
+ "kafka_write notification Exception {}".format(e
)
6047 self
.logger
.debug(logging_text
+ "Exit")
6048 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6050 async def _scale_kdu(
6051 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6053 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6054 for kdu_name
in _scaling_info
:
6055 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6056 deployed_kdu
, index
= get_deployed_kdu(
6057 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6059 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6060 kdu_instance
= deployed_kdu
["kdu-instance"]
6061 scale
= int(kdu_scaling_info
["scale"])
6062 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6065 "collection": "nsrs",
6066 "filter": {"_id": nsr_id
},
6067 "path": "_admin.deployed.K8s.{}".format(index
),
6070 step
= "scaling application {}".format(
6071 kdu_scaling_info
["resource-name"]
6073 self
.logger
.debug(logging_text
+ step
)
6075 if kdu_scaling_info
["type"] == "delete":
6076 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6079 and kdu_config
.get("terminate-config-primitive")
6080 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6082 terminate_config_primitive_list
= kdu_config
.get(
6083 "terminate-config-primitive"
6085 terminate_config_primitive_list
.sort(
6086 key
=lambda val
: int(val
["seq"])
6090 terminate_config_primitive
6091 ) in terminate_config_primitive_list
:
6092 primitive_params_
= self
._map
_primitive
_params
(
6093 terminate_config_primitive
, {}, {}
6095 step
= "execute terminate config primitive"
6096 self
.logger
.debug(logging_text
+ step
)
6097 await asyncio
.wait_for(
6098 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6099 cluster_uuid
=cluster_uuid
,
6100 kdu_instance
=kdu_instance
,
6101 primitive_name
=terminate_config_primitive
["name"],
6102 params
=primitive_params_
,
6109 await asyncio
.wait_for(
6110 self
.k8scluster_map
[k8s_cluster_type
].scale(
6113 kdu_scaling_info
["resource-name"],
6116 timeout
=self
.timeout_vca_on_error
,
6119 if kdu_scaling_info
["type"] == "create":
6120 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6123 and kdu_config
.get("initial-config-primitive")
6124 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6126 initial_config_primitive_list
= kdu_config
.get(
6127 "initial-config-primitive"
6129 initial_config_primitive_list
.sort(
6130 key
=lambda val
: int(val
["seq"])
6133 for initial_config_primitive
in initial_config_primitive_list
:
6134 primitive_params_
= self
._map
_primitive
_params
(
6135 initial_config_primitive
, {}, {}
6137 step
= "execute initial config primitive"
6138 self
.logger
.debug(logging_text
+ step
)
6139 await asyncio
.wait_for(
6140 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6141 cluster_uuid
=cluster_uuid
,
6142 kdu_instance
=kdu_instance
,
6143 primitive_name
=initial_config_primitive
["name"],
6144 params
=primitive_params_
,
6151 async def _scale_ng_ro(
6152 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6154 nsr_id
= db_nslcmop
["nsInstanceId"]
6155 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6158 # read from db: vnfd's for every vnf
6161 # for each vnf in ns, read vnfd
6162 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6163 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6164 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6165 # if we haven't this vnfd, read it from db
6166 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6168 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6169 db_vnfds
.append(vnfd
)
6170 n2vc_key
= self
.n2vc
.get_public_key()
6171 n2vc_key_list
= [n2vc_key
]
6174 vdu_scaling_info
.get("vdu-create"),
6175 vdu_scaling_info
.get("vdu-delete"),
6178 # db_vnfr has been updated, update db_vnfrs to use it
6179 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6180 await self
._instantiate
_ng
_ro
(
6190 start_deploy
=time(),
6191 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6193 if vdu_scaling_info
.get("vdu-delete"):
6195 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6198 async def add_prometheus_metrics(
6199 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6201 if not self
.prometheus
:
6203 # look if exist a file called 'prometheus*.j2' and
6204 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6208 for f
in artifact_content
6209 if f
.startswith("prometheus") and f
.endswith(".j2")
6215 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6219 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6220 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6222 vnfr_id
= vnfr_id
.replace("-", "")
6224 "JOB_NAME": vnfr_id
,
6225 "TARGET_IP": target_ip
,
6226 "EXPORTER_POD_IP": host_name
,
6227 "EXPORTER_POD_PORT": host_port
,
6229 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
6230 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6231 for job
in job_list
:
6233 not isinstance(job
.get("job_name"), str)
6234 or vnfr_id
not in job
["job_name"]
6236 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6237 job
["nsr_id"] = nsr_id
6238 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
6239 if await self
.prometheus
.update(job_dict
):
6240 return list(job_dict
.keys())
6242 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6244 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6246 :param: vim_account_id: VIM Account ID
6248 :return: (cloud_name, cloud_credential)
6250 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6251 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6253 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6255 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6257 :param: vim_account_id: VIM Account ID
6259 :return: (cloud_name, cloud_credential)
6261 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6262 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")