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 nslcmop_id
= db_nslcmop
["_id"]
801 "name": db_nsr
["name"],
804 "image": deepcopy(db_nsr
["image"]),
805 "flavor": deepcopy(db_nsr
["flavor"]),
806 "action_id": nslcmop_id
,
807 "cloud_init_content": {},
809 for image
in target
["image"]:
810 image
["vim_info"] = {}
811 for flavor
in target
["flavor"]:
812 flavor
["vim_info"] = {}
814 if db_nslcmop
.get("lcmOperationType") != "instantiate":
815 # get parameters of instantiation:
816 db_nslcmop_instantiate
= self
.db
.get_list(
819 "nsInstanceId": db_nslcmop
["nsInstanceId"],
820 "lcmOperationType": "instantiate",
823 ns_params
= db_nslcmop_instantiate
.get("operationParams")
825 ns_params
= db_nslcmop
.get("operationParams")
826 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
827 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
830 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
831 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
835 "mgmt-network": vld
.get("mgmt-network", False),
836 "type": vld
.get("type"),
839 "vim_network_name": vld
.get("vim-network-name"),
840 "vim_account_id": ns_params
["vimAccountId"],
844 # check if this network needs SDN assist
845 if vld
.get("pci-interfaces"):
846 db_vim
= get_vim_account(ns_params
["vimAccountId"])
847 sdnc_id
= db_vim
["config"].get("sdn-controller")
849 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
850 target_sdn
= "sdn:{}".format(sdnc_id
)
851 target_vld
["vim_info"][target_sdn
] = {
853 "target_vim": target_vim
,
855 "type": vld
.get("type"),
858 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
859 for nsd_vnf_profile
in nsd_vnf_profiles
:
860 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
861 if cp
["virtual-link-profile-id"] == vld
["id"]:
863 "member_vnf:{}.{}".format(
864 cp
["constituent-cpd-id"][0][
865 "constituent-base-element-id"
867 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
869 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
871 # check at nsd descriptor, if there is an ip-profile
873 nsd_vlp
= find_in_list(
874 get_virtual_link_profiles(nsd
),
875 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
880 and nsd_vlp
.get("virtual-link-protocol-data")
881 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
883 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
886 ip_profile_dest_data
= {}
887 if "ip-version" in ip_profile_source_data
:
888 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
891 if "cidr" in ip_profile_source_data
:
892 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
895 if "gateway-ip" in ip_profile_source_data
:
896 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
899 if "dhcp-enabled" in ip_profile_source_data
:
900 ip_profile_dest_data
["dhcp-params"] = {
901 "enabled": ip_profile_source_data
["dhcp-enabled"]
903 vld_params
["ip-profile"] = ip_profile_dest_data
905 # update vld_params with instantiation params
906 vld_instantiation_params
= find_in_list(
907 get_iterable(ns_params
, "vld"),
908 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
910 if vld_instantiation_params
:
911 vld_params
.update(vld_instantiation_params
)
912 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
913 target
["ns"]["vld"].append(target_vld
)
915 for vnfr
in db_vnfrs
.values():
917 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
919 vnf_params
= find_in_list(
920 get_iterable(ns_params
, "vnf"),
921 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
923 target_vnf
= deepcopy(vnfr
)
924 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
925 for vld
in target_vnf
.get("vld", ()):
926 # check if connected to a ns.vld, to fill target'
927 vnf_cp
= find_in_list(
928 vnfd
.get("int-virtual-link-desc", ()),
929 lambda cpd
: cpd
.get("id") == vld
["id"],
932 ns_cp
= "member_vnf:{}.{}".format(
933 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
935 if cp2target
.get(ns_cp
):
936 vld
["target"] = cp2target
[ns_cp
]
939 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
941 # check if this network needs SDN assist
943 if vld
.get("pci-interfaces"):
944 db_vim
= get_vim_account(vnfr
["vim-account-id"])
945 sdnc_id
= db_vim
["config"].get("sdn-controller")
947 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
948 target_sdn
= "sdn:{}".format(sdnc_id
)
949 vld
["vim_info"][target_sdn
] = {
951 "target_vim": target_vim
,
953 "type": vld
.get("type"),
956 # check at vnfd descriptor, if there is an ip-profile
958 vnfd_vlp
= find_in_list(
959 get_virtual_link_profiles(vnfd
),
960 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
964 and vnfd_vlp
.get("virtual-link-protocol-data")
965 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
967 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
970 ip_profile_dest_data
= {}
971 if "ip-version" in ip_profile_source_data
:
972 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
975 if "cidr" in ip_profile_source_data
:
976 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
979 if "gateway-ip" in ip_profile_source_data
:
980 ip_profile_dest_data
[
982 ] = ip_profile_source_data
["gateway-ip"]
983 if "dhcp-enabled" in ip_profile_source_data
:
984 ip_profile_dest_data
["dhcp-params"] = {
985 "enabled": ip_profile_source_data
["dhcp-enabled"]
988 vld_params
["ip-profile"] = ip_profile_dest_data
989 # update vld_params with instantiation params
991 vld_instantiation_params
= find_in_list(
992 get_iterable(vnf_params
, "internal-vld"),
993 lambda i_vld
: i_vld
["name"] == vld
["id"],
995 if vld_instantiation_params
:
996 vld_params
.update(vld_instantiation_params
)
997 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1000 for vdur
in target_vnf
.get("vdur", ()):
1001 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1002 continue # This vdu must not be created
1003 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1005 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1008 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1009 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1012 and vdu_configuration
.get("config-access")
1013 and vdu_configuration
.get("config-access").get("ssh-access")
1015 vdur
["ssh-keys"] = ssh_keys_all
1016 vdur
["ssh-access-required"] = vdu_configuration
[
1018 ]["ssh-access"]["required"]
1021 and vnf_configuration
.get("config-access")
1022 and vnf_configuration
.get("config-access").get("ssh-access")
1023 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1025 vdur
["ssh-keys"] = ssh_keys_all
1026 vdur
["ssh-access-required"] = vnf_configuration
[
1028 ]["ssh-access"]["required"]
1029 elif ssh_keys_instantiation
and find_in_list(
1030 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1032 vdur
["ssh-keys"] = ssh_keys_instantiation
1034 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1036 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1038 if vdud
.get("cloud-init-file"):
1039 vdur
["cloud-init"] = "{}:file:{}".format(
1040 vnfd
["_id"], vdud
.get("cloud-init-file")
1042 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1043 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1044 base_folder
= vnfd
["_admin"]["storage"]
1045 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1046 base_folder
["folder"],
1047 base_folder
["pkg-dir"],
1048 vdud
.get("cloud-init-file"),
1050 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
1051 target
["cloud_init_content"][
1054 elif vdud
.get("cloud-init"):
1055 vdur
["cloud-init"] = "{}:vdu:{}".format(
1056 vnfd
["_id"], get_vdu_index(vnfd
, vdur
["vdu-id-ref"])
1058 # put content at target.cloul_init_content. Avoid ng_ro read vnfd descriptor
1059 target
["cloud_init_content"][vdur
["cloud-init"]] = vdud
[
1062 vdur
["additionalParams"] = vdur
.get("additionalParams") or {}
1063 deploy_params_vdu
= self
._format
_additional
_params
(
1064 vdur
.get("additionalParams") or {}
1066 deploy_params_vdu
["OSM"] = get_osm_params(
1067 vnfr
, vdur
["vdu-id-ref"], vdur
["count-index"]
1069 vdur
["additionalParams"] = deploy_params_vdu
1072 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
1073 if target_vim
not in ns_flavor
["vim_info"]:
1074 ns_flavor
["vim_info"][target_vim
] = {}
1077 # in case alternative images are provided we must check if they should be applied
1078 # for the vim_type, modify the vim_type taking into account
1079 ns_image_id
= int(vdur
["ns-image-id"])
1080 if vdur
.get("alt-image-ids"):
1081 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1082 vim_type
= db_vim
["vim_type"]
1083 for alt_image_id
in vdur
.get("alt-image-ids"):
1084 ns_alt_image
= target
["image"][int(alt_image_id
)]
1085 if vim_type
== ns_alt_image
.get("vim-type"):
1086 # must use alternative image
1088 "use alternative image id: {}".format(alt_image_id
)
1090 ns_image_id
= alt_image_id
1091 vdur
["ns-image-id"] = ns_image_id
1093 ns_image
= target
["image"][int(ns_image_id
)]
1094 if target_vim
not in ns_image
["vim_info"]:
1095 ns_image
["vim_info"][target_vim
] = {}
1097 vdur
["vim_info"] = {target_vim
: {}}
1098 # instantiation parameters
1100 # vdu_instantiation_params = next((v for v in get_iterable(vnf_params, "vdu") if v["id"] ==
1101 # vdud["id"]), None)
1102 vdur_list
.append(vdur
)
1103 target_vnf
["vdur"] = vdur_list
1104 target
["vnf"].append(target_vnf
)
1106 desc
= await self
.RO
.deploy(nsr_id
, target
)
1107 self
.logger
.debug("RO return > {}".format(desc
))
1108 action_id
= desc
["action_id"]
1109 await self
._wait
_ng
_ro
(
1110 nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
1115 "_admin.deployed.RO.operational-status": "running",
1116 "detailed-status": " ".join(stage
),
1118 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1119 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1120 self
._write
_op
_status
(nslcmop_id
, stage
)
1122 logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
)
1126 async def _wait_ng_ro(
1135 detailed_status_old
= None
1137 start_time
= start_time
or time()
1138 while time() <= start_time
+ timeout
:
1139 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
1140 self
.logger
.debug("Wait NG RO > {}".format(desc_status
))
1141 if desc_status
["status"] == "FAILED":
1142 raise NgRoException(desc_status
["details"])
1143 elif desc_status
["status"] == "BUILD":
1145 stage
[2] = "VIM: ({})".format(desc_status
["details"])
1146 elif desc_status
["status"] == "DONE":
1148 stage
[2] = "Deployed at VIM"
1151 assert False, "ROclient.check_ns_status returns unknown {}".format(
1152 desc_status
["status"]
1154 if stage
and nslcmop_id
and stage
[2] != detailed_status_old
:
1155 detailed_status_old
= stage
[2]
1156 db_nsr_update
["detailed-status"] = " ".join(stage
)
1157 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1158 self
._write
_op
_status
(nslcmop_id
, stage
)
1159 await asyncio
.sleep(15, loop
=self
.loop
)
1160 else: # timeout_ns_deploy
1161 raise NgRoException("Timeout waiting ns to deploy")
1163 async def _terminate_ng_ro(
1164 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
1169 start_deploy
= time()
1176 "action_id": nslcmop_id
,
1178 desc
= await self
.RO
.deploy(nsr_id
, target
)
1179 action_id
= desc
["action_id"]
1180 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1181 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1184 + "ns terminate action at RO. action_id={}".format(action_id
)
1188 delete_timeout
= 20 * 60 # 20 minutes
1189 await self
._wait
_ng
_ro
(
1190 nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
1193 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1194 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1196 await self
.RO
.delete(nsr_id
)
1197 except Exception as e
:
1198 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1199 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1200 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1201 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1203 logging_text
+ "RO_action_id={} already deleted".format(action_id
)
1205 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1206 failed_detail
.append("delete conflict: {}".format(e
))
1209 + "RO_action_id={} delete conflict: {}".format(action_id
, e
)
1212 failed_detail
.append("delete error: {}".format(e
))
1215 + "RO_action_id={} delete error: {}".format(action_id
, e
)
1219 stage
[2] = "Error deleting from VIM"
1221 stage
[2] = "Deleted from VIM"
1222 db_nsr_update
["detailed-status"] = " ".join(stage
)
1223 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1224 self
._write
_op
_status
(nslcmop_id
, stage
)
1227 raise LcmException("; ".join(failed_detail
))
1230 async def instantiate_RO(
1244 :param logging_text: preffix text to use at logging
1245 :param nsr_id: nsr identity
1246 :param nsd: database content of ns descriptor
1247 :param db_nsr: database content of ns record
1248 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1250 :param db_vnfds: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1251 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1252 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1253 :return: None or exception
1256 start_deploy
= time()
1257 ns_params
= db_nslcmop
.get("operationParams")
1258 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1259 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1261 timeout_ns_deploy
= self
.timeout
.get(
1262 "ns_deploy", self
.timeout_ns_deploy
1265 # Check for and optionally request placement optimization. Database will be updated if placement activated
1266 stage
[2] = "Waiting for Placement."
1267 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1268 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1269 for vnfr
in db_vnfrs
.values():
1270 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1273 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1275 return await self
._instantiate
_ng
_ro
(
1288 except Exception as e
:
1289 stage
[2] = "ERROR deploying at VIM"
1290 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1292 "Error deploying at VIM {}".format(e
),
1293 exc_info
=not isinstance(
1296 ROclient
.ROClientException
,
1305 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1307 Wait for kdu to be up, get ip address
1308 :param logging_text: prefix use for logging
1315 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1318 while nb_tries
< 360:
1319 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1323 for x
in get_iterable(db_vnfr
, "kdur")
1324 if x
.get("kdu-name") == kdu_name
1330 "Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
)
1332 if kdur
.get("status"):
1333 if kdur
["status"] in ("READY", "ENABLED"):
1334 return kdur
.get("ip-address")
1337 "target KDU={} is in error state".format(kdu_name
)
1340 await asyncio
.sleep(10, loop
=self
.loop
)
1342 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1344 async def wait_vm_up_insert_key_ro(
1345 self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None
1348 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1349 :param logging_text: prefix use for logging
1354 :param pub_key: public ssh key to inject, None to skip
1355 :param user: user to apply the public ssh key
1359 self
.logger
.debug(logging_text
+ "Starting wait_vm_up_insert_key_ro")
1363 target_vdu_id
= None
1369 if ro_retries
>= 360: # 1 hour
1371 "Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
)
1374 await asyncio
.sleep(10, loop
=self
.loop
)
1377 if not target_vdu_id
:
1378 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1380 if not vdu_id
: # for the VNF case
1381 if db_vnfr
.get("status") == "ERROR":
1383 "Cannot inject ssh-key because target VNF is in error state"
1385 ip_address
= db_vnfr
.get("ip-address")
1391 for x
in get_iterable(db_vnfr
, "vdur")
1392 if x
.get("ip-address") == ip_address
1400 for x
in get_iterable(db_vnfr
, "vdur")
1401 if x
.get("vdu-id-ref") == vdu_id
1402 and x
.get("count-index") == vdu_index
1408 not vdur
and len(db_vnfr
.get("vdur", ())) == 1
1409 ): # If only one, this should be the target vdu
1410 vdur
= db_vnfr
["vdur"][0]
1413 "Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(
1414 vnfr_id
, vdu_id
, vdu_index
1417 # New generation RO stores information at "vim_info"
1420 if vdur
.get("vim_info"):
1422 t
for t
in vdur
["vim_info"]
1423 ) # there should be only one key
1424 ng_ro_status
= vdur
["vim_info"][target_vim
].get("vim_status")
1426 vdur
.get("pdu-type")
1427 or vdur
.get("status") == "ACTIVE"
1428 or ng_ro_status
== "ACTIVE"
1430 ip_address
= vdur
.get("ip-address")
1433 target_vdu_id
= vdur
["vdu-id-ref"]
1434 elif vdur
.get("status") == "ERROR" or ng_ro_status
== "ERROR":
1436 "Cannot inject ssh-key because target VM is in error state"
1439 if not target_vdu_id
:
1442 # inject public key into machine
1443 if pub_key
and user
:
1444 self
.logger
.debug(logging_text
+ "Inserting RO key")
1445 self
.logger
.debug("SSH > PubKey > {}".format(pub_key
))
1446 if vdur
.get("pdu-type"):
1447 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1450 ro_vm_id
= "{}-{}".format(
1451 db_vnfr
["member-vnf-index-ref"], target_vdu_id
1452 ) # TODO add vdu_index
1456 "action": "inject_ssh_key",
1460 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdur
["id"]}]}],
1462 desc
= await self
.RO
.deploy(nsr_id
, target
)
1463 action_id
= desc
["action_id"]
1464 await self
._wait
_ng
_ro
(nsr_id
, action_id
, timeout
=600)
1467 # wait until NS is deployed at RO
1469 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1470 ro_nsr_id
= deep_get(
1471 db_nsrs
, ("_admin", "deployed", "RO", "nsr_id")
1475 result_dict
= await self
.RO
.create_action(
1477 item_id_name
=ro_nsr_id
,
1479 "add_public_key": pub_key
,
1484 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1485 if not result_dict
or not isinstance(result_dict
, dict):
1487 "Unknown response from RO when injecting key"
1489 for result
in result_dict
.values():
1490 if result
.get("vim_result") == 200:
1493 raise ROclient
.ROClientException(
1494 "error injecting key: {}".format(
1495 result
.get("description")
1499 except NgRoException
as e
:
1501 "Reaching max tries injecting key. Error: {}".format(e
)
1503 except ROclient
.ROClientException
as e
:
1507 + "error injecting key: {}. Retrying until {} seconds".format(
1514 "Reaching max tries injecting key. Error: {}".format(e
)
1521 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1523 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1525 my_vca
= vca_deployed_list
[vca_index
]
1526 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1527 # vdu or kdu: no dependencies
1531 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1532 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1533 configuration_status_list
= db_nsr
["configurationStatus"]
1534 for index
, vca_deployed
in enumerate(configuration_status_list
):
1535 if index
== vca_index
:
1538 if not my_vca
.get("member-vnf-index") or (
1539 vca_deployed
.get("member-vnf-index")
1540 == my_vca
.get("member-vnf-index")
1542 internal_status
= configuration_status_list
[index
].get("status")
1543 if internal_status
== "READY":
1545 elif internal_status
== "BROKEN":
1547 "Configuration aborted because dependent charm/s has failed"
1552 # no dependencies, return
1554 await asyncio
.sleep(10)
1557 raise LcmException("Configuration aborted because dependent charm/s timeout")
1559 def get_vca_id(self
, db_vnfr
: dict, db_nsr
: dict):
1560 return deep_get(db_vnfr
, ("vca-id",)) or deep_get(
1561 db_nsr
, ("instantiate_params", "vcaId")
1564 async def instantiate_N2VC(
1581 ee_config_descriptor
,
1583 nsr_id
= db_nsr
["_id"]
1584 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1585 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1586 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1587 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1589 "collection": "nsrs",
1590 "filter": {"_id": nsr_id
},
1591 "path": db_update_entry
,
1597 element_under_configuration
= nsr_id
1601 vnfr_id
= db_vnfr
["_id"]
1602 osm_config
["osm"]["vnf_id"] = vnfr_id
1604 namespace
= "{nsi}.{ns}".format(nsi
=nsi_id
if nsi_id
else "", ns
=nsr_id
)
1607 element_type
= "VNF"
1608 element_under_configuration
= vnfr_id
1609 namespace
+= ".{}-{}".format(vnfr_id
, vdu_index
or 0)
1611 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1612 element_type
= "VDU"
1613 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1614 osm_config
["osm"]["vdu_id"] = vdu_id
1616 namespace
+= ".{}.{}".format(kdu_name
, vdu_index
or 0)
1617 element_type
= "KDU"
1618 element_under_configuration
= kdu_name
1619 osm_config
["osm"]["kdu_name"] = kdu_name
1622 artifact_path
= "{}/{}/{}/{}".format(
1623 base_folder
["folder"],
1624 base_folder
["pkg-dir"],
1626 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1631 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1633 # get initial_config_primitive_list that applies to this element
1634 initial_config_primitive_list
= config_descriptor
.get(
1635 "initial-config-primitive"
1639 "Initial config primitive list > {}".format(
1640 initial_config_primitive_list
1644 # add config if not present for NS charm
1645 ee_descriptor_id
= ee_config_descriptor
.get("id")
1646 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1647 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1648 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1652 "Initial config primitive list #2 > {}".format(
1653 initial_config_primitive_list
1656 # n2vc_redesign STEP 3.1
1657 # find old ee_id if exists
1658 ee_id
= vca_deployed
.get("ee_id")
1660 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1661 # create or register execution environment in VCA
1662 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1664 self
._write
_configuration
_status
(
1666 vca_index
=vca_index
,
1668 element_under_configuration
=element_under_configuration
,
1669 element_type
=element_type
,
1672 step
= "create execution environment"
1673 self
.logger
.debug(logging_text
+ step
)
1677 if vca_type
== "k8s_proxy_charm":
1678 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1679 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1680 namespace
=namespace
,
1681 artifact_path
=artifact_path
,
1685 elif vca_type
== "helm" or vca_type
== "helm-v3":
1686 ee_id
, credentials
= await self
.vca_map
[
1688 ].create_execution_environment(
1689 namespace
=namespace
,
1693 artifact_path
=artifact_path
,
1697 ee_id
, credentials
= await self
.vca_map
[
1699 ].create_execution_environment(
1700 namespace
=namespace
,
1706 elif vca_type
== "native_charm":
1707 step
= "Waiting to VM being up and getting IP address"
1708 self
.logger
.debug(logging_text
+ step
)
1709 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1718 credentials
= {"hostname": rw_mgmt_ip
}
1720 username
= deep_get(
1721 config_descriptor
, ("config-access", "ssh-access", "default-user")
1723 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1724 # merged. Meanwhile let's get username from initial-config-primitive
1725 if not username
and initial_config_primitive_list
:
1726 for config_primitive
in initial_config_primitive_list
:
1727 for param
in config_primitive
.get("parameter", ()):
1728 if param
["name"] == "ssh-username":
1729 username
= param
["value"]
1733 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1734 "'config-access.ssh-access.default-user'"
1736 credentials
["username"] = username
1737 # n2vc_redesign STEP 3.2
1739 self
._write
_configuration
_status
(
1741 vca_index
=vca_index
,
1742 status
="REGISTERING",
1743 element_under_configuration
=element_under_configuration
,
1744 element_type
=element_type
,
1747 step
= "register execution environment {}".format(credentials
)
1748 self
.logger
.debug(logging_text
+ step
)
1749 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1750 credentials
=credentials
,
1751 namespace
=namespace
,
1756 # for compatibility with MON/POL modules, the need model and application name at database
1757 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1758 ee_id_parts
= ee_id
.split(".")
1759 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1760 if len(ee_id_parts
) >= 2:
1761 model_name
= ee_id_parts
[0]
1762 application_name
= ee_id_parts
[1]
1763 db_nsr_update
[db_update_entry
+ "model"] = model_name
1764 db_nsr_update
[db_update_entry
+ "application"] = application_name
1766 # n2vc_redesign STEP 3.3
1767 step
= "Install configuration Software"
1769 self
._write
_configuration
_status
(
1771 vca_index
=vca_index
,
1772 status
="INSTALLING SW",
1773 element_under_configuration
=element_under_configuration
,
1774 element_type
=element_type
,
1775 other_update
=db_nsr_update
,
1778 # TODO check if already done
1779 self
.logger
.debug(logging_text
+ step
)
1781 if vca_type
== "native_charm":
1782 config_primitive
= next(
1783 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1786 if config_primitive
:
1787 config
= self
._map
_primitive
_params
(
1788 config_primitive
, {}, deploy_params
1791 if vca_type
== "lxc_proxy_charm":
1792 if element_type
== "NS":
1793 num_units
= db_nsr
.get("config-units") or 1
1794 elif element_type
== "VNF":
1795 num_units
= db_vnfr
.get("config-units") or 1
1796 elif element_type
== "VDU":
1797 for v
in db_vnfr
["vdur"]:
1798 if vdu_id
== v
["vdu-id-ref"]:
1799 num_units
= v
.get("config-units") or 1
1801 if vca_type
!= "k8s_proxy_charm":
1802 await self
.vca_map
[vca_type
].install_configuration_sw(
1804 artifact_path
=artifact_path
,
1807 num_units
=num_units
,
1811 # write in db flag of configuration_sw already installed
1813 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1816 # add relations for this VCA (wait for other peers related with this VCA)
1817 await self
._add
_vca
_relations
(
1818 logging_text
=logging_text
,
1820 vca_index
=vca_index
,
1825 # if SSH access is required, then get execution environment SSH public
1826 # if native charm we have waited already to VM be UP
1827 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1830 # self.logger.debug("get ssh key block")
1832 config_descriptor
, ("config-access", "ssh-access", "required")
1834 # self.logger.debug("ssh key needed")
1835 # Needed to inject a ssh key
1838 ("config-access", "ssh-access", "default-user"),
1840 step
= "Install configuration Software, getting public ssh key"
1841 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1842 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1845 step
= "Insert public key into VM user={} ssh_key={}".format(
1849 # self.logger.debug("no need to get ssh key")
1850 step
= "Waiting to VM being up and getting IP address"
1851 self
.logger
.debug(logging_text
+ step
)
1853 # n2vc_redesign STEP 5.1
1854 # wait for RO (ip-address) Insert pub_key into VM
1857 rw_mgmt_ip
= await self
.wait_kdu_up(
1858 logging_text
, nsr_id
, vnfr_id
, kdu_name
1861 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1871 rw_mgmt_ip
= None # This is for a NS configuration
1873 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1875 # store rw_mgmt_ip in deploy params for later replacement
1876 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1878 # n2vc_redesign STEP 6 Execute initial config primitive
1879 step
= "execute initial config primitive"
1881 # wait for dependent primitives execution (NS -> VNF -> VDU)
1882 if initial_config_primitive_list
:
1883 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1885 # stage, in function of element type: vdu, kdu, vnf or ns
1886 my_vca
= vca_deployed_list
[vca_index
]
1887 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1889 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1890 elif my_vca
.get("member-vnf-index"):
1892 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1895 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1897 self
._write
_configuration
_status
(
1898 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1901 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1903 check_if_terminated_needed
= True
1904 for initial_config_primitive
in initial_config_primitive_list
:
1905 # adding information on the vca_deployed if it is a NS execution environment
1906 if not vca_deployed
["member-vnf-index"]:
1907 deploy_params
["ns_config_info"] = json
.dumps(
1908 self
._get
_ns
_config
_info
(nsr_id
)
1910 # TODO check if already done
1911 primitive_params_
= self
._map
_primitive
_params
(
1912 initial_config_primitive
, {}, deploy_params
1915 step
= "execute primitive '{}' params '{}'".format(
1916 initial_config_primitive
["name"], primitive_params_
1918 self
.logger
.debug(logging_text
+ step
)
1919 await self
.vca_map
[vca_type
].exec_primitive(
1921 primitive_name
=initial_config_primitive
["name"],
1922 params_dict
=primitive_params_
,
1926 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1927 if check_if_terminated_needed
:
1928 if config_descriptor
.get("terminate-config-primitive"):
1930 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
1932 check_if_terminated_needed
= False
1934 # TODO register in database that primitive is done
1936 # STEP 7 Configure metrics
1937 if vca_type
== "helm" or vca_type
== "helm-v3":
1938 prometheus_jobs
= await self
.add_prometheus_metrics(
1940 artifact_path
=artifact_path
,
1941 ee_config_descriptor
=ee_config_descriptor
,
1944 target_ip
=rw_mgmt_ip
,
1950 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
1953 step
= "instantiated at VCA"
1954 self
.logger
.debug(logging_text
+ step
)
1956 self
._write
_configuration
_status
(
1957 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
1960 except Exception as e
: # TODO not use Exception but N2VC exception
1961 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1963 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
1966 "Exception while {} : {}".format(step
, e
), exc_info
=True
1968 self
._write
_configuration
_status
(
1969 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
1971 raise LcmException("{} {}".format(step
, e
)) from e
1973 def _write_ns_status(
1977 current_operation
: str,
1978 current_operation_id
: str,
1979 error_description
: str = None,
1980 error_detail
: str = None,
1981 other_update
: dict = None,
1984 Update db_nsr fields.
1987 :param current_operation:
1988 :param current_operation_id:
1989 :param error_description:
1990 :param error_detail:
1991 :param other_update: Other required changes at database if provided, will be cleared
1995 db_dict
= other_update
or {}
1998 ] = current_operation_id
# for backward compatibility
1999 db_dict
["_admin.current-operation"] = current_operation_id
2000 db_dict
["_admin.operation-type"] = (
2001 current_operation
if current_operation
!= "IDLE" else None
2003 db_dict
["currentOperation"] = current_operation
2004 db_dict
["currentOperationID"] = current_operation_id
2005 db_dict
["errorDescription"] = error_description
2006 db_dict
["errorDetail"] = error_detail
2009 db_dict
["nsState"] = ns_state
2010 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2011 except DbException
as e
:
2012 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2014 def _write_op_status(
2018 error_message
: str = None,
2019 queuePosition
: int = 0,
2020 operation_state
: str = None,
2021 other_update
: dict = None,
2024 db_dict
= other_update
or {}
2025 db_dict
["queuePosition"] = queuePosition
2026 if isinstance(stage
, list):
2027 db_dict
["stage"] = stage
[0]
2028 db_dict
["detailed-status"] = " ".join(stage
)
2029 elif stage
is not None:
2030 db_dict
["stage"] = str(stage
)
2032 if error_message
is not None:
2033 db_dict
["errorMessage"] = error_message
2034 if operation_state
is not None:
2035 db_dict
["operationState"] = operation_state
2036 db_dict
["statusEnteredTime"] = time()
2037 self
.update_db_2("nslcmops", op_id
, db_dict
)
2038 except DbException
as e
:
2040 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2043 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2045 nsr_id
= db_nsr
["_id"]
2046 # configurationStatus
2047 config_status
= db_nsr
.get("configurationStatus")
2050 "configurationStatus.{}.status".format(index
): status
2051 for index
, v
in enumerate(config_status
)
2055 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2057 except DbException
as e
:
2059 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2062 def _write_configuration_status(
2067 element_under_configuration
: str = None,
2068 element_type
: str = None,
2069 other_update
: dict = None,
2072 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2073 # .format(vca_index, status))
2076 db_path
= "configurationStatus.{}.".format(vca_index
)
2077 db_dict
= other_update
or {}
2079 db_dict
[db_path
+ "status"] = status
2080 if element_under_configuration
:
2082 db_path
+ "elementUnderConfiguration"
2083 ] = element_under_configuration
2085 db_dict
[db_path
+ "elementType"] = element_type
2086 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2087 except DbException
as e
:
2089 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2090 status
, nsr_id
, vca_index
, e
2094 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2096 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2097 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2098 Database is used because the result can be obtained from a different LCM worker in case of HA.
2099 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2100 :param db_nslcmop: database content of nslcmop
2101 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2102 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2103 computed 'vim-account-id'
2106 nslcmop_id
= db_nslcmop
["_id"]
2107 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2108 if placement_engine
== "PLA":
2110 logging_text
+ "Invoke and wait for placement optimization"
2112 await self
.msg
.aiowrite(
2113 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2115 db_poll_interval
= 5
2116 wait
= db_poll_interval
* 10
2118 while not pla_result
and wait
>= 0:
2119 await asyncio
.sleep(db_poll_interval
)
2120 wait
-= db_poll_interval
2121 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2122 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2126 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2129 for pla_vnf
in pla_result
["vnf"]:
2130 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2131 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2136 {"_id": vnfr
["_id"]},
2137 {"vim-account-id": pla_vnf
["vimAccountId"]},
2140 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2143 def update_nsrs_with_pla_result(self
, params
):
2145 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2147 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2149 except Exception as e
:
2150 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2152 async def instantiate(self
, nsr_id
, nslcmop_id
):
2155 :param nsr_id: ns instance to deploy
2156 :param nslcmop_id: operation to run
2160 # Try to lock HA task here
2161 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2162 if not task_is_locked_by_me
:
2164 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2168 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2169 self
.logger
.debug(logging_text
+ "Enter")
2171 # get all needed from database
2173 # database nsrs record
2176 # database nslcmops record
2179 # update operation on nsrs
2181 # update operation on nslcmops
2182 db_nslcmop_update
= {}
2184 nslcmop_operation_state
= None
2185 db_vnfrs
= {} # vnf's info indexed by member-index
2187 tasks_dict_info
= {} # from task to info text
2191 "Stage 1/5: preparation of the environment.",
2192 "Waiting for previous operations to terminate.",
2195 # ^ stage, step, VIM progress
2197 # wait for any previous tasks in process
2198 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2200 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2201 stage
[1] = "Reading from database."
2202 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2203 db_nsr_update
["detailed-status"] = "creating"
2204 db_nsr_update
["operational-status"] = "init"
2205 self
._write
_ns
_status
(
2207 ns_state
="BUILDING",
2208 current_operation
="INSTANTIATING",
2209 current_operation_id
=nslcmop_id
,
2210 other_update
=db_nsr_update
,
2212 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2214 # read from db: operation
2215 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2216 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2217 ns_params
= db_nslcmop
.get("operationParams")
2218 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2219 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2221 timeout_ns_deploy
= self
.timeout
.get(
2222 "ns_deploy", self
.timeout_ns_deploy
2226 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2227 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2228 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2229 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2230 self
.fs
.sync(db_nsr
["nsd-id"])
2232 # nsr_name = db_nsr["name"] # TODO short-name??
2234 # read from db: vnf's of this ns
2235 stage
[1] = "Getting vnfrs from db."
2236 self
.logger
.debug(logging_text
+ stage
[1])
2237 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2239 # read from db: vnfd's for every vnf
2240 db_vnfds
= [] # every vnfd data
2242 # for each vnf in ns, read vnfd
2243 for vnfr
in db_vnfrs_list
:
2244 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2245 vnfd_id
= vnfr
["vnfd-id"]
2246 vnfd_ref
= vnfr
["vnfd-ref"]
2247 self
.fs
.sync(vnfd_id
)
2249 # if we haven't this vnfd, read it from db
2250 if vnfd_id
not in db_vnfds
:
2252 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2255 self
.logger
.debug(logging_text
+ stage
[1])
2256 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2259 db_vnfds
.append(vnfd
)
2261 # Get or generates the _admin.deployed.VCA list
2262 vca_deployed_list
= None
2263 if db_nsr
["_admin"].get("deployed"):
2264 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2265 if vca_deployed_list
is None:
2266 vca_deployed_list
= []
2267 configuration_status_list
= []
2268 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2269 db_nsr_update
["configurationStatus"] = configuration_status_list
2270 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2271 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2272 elif isinstance(vca_deployed_list
, dict):
2273 # maintain backward compatibility. Change a dict to list at database
2274 vca_deployed_list
= list(vca_deployed_list
.values())
2275 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2276 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2279 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2281 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2282 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2284 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2285 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2286 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2288 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2291 # n2vc_redesign STEP 2 Deploy Network Scenario
2292 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2293 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2295 stage
[1] = "Deploying KDUs."
2296 # self.logger.debug(logging_text + "Before deploy_kdus")
2297 # Call to deploy_kdus in case exists the "vdu:kdu" param
2298 await self
.deploy_kdus(
2299 logging_text
=logging_text
,
2301 nslcmop_id
=nslcmop_id
,
2304 task_instantiation_info
=tasks_dict_info
,
2307 stage
[1] = "Getting VCA public key."
2308 # n2vc_redesign STEP 1 Get VCA public ssh-key
2309 # feature 1429. Add n2vc public key to needed VMs
2310 n2vc_key
= self
.n2vc
.get_public_key()
2311 n2vc_key_list
= [n2vc_key
]
2312 if self
.vca_config
.get("public_key"):
2313 n2vc_key_list
.append(self
.vca_config
["public_key"])
2315 stage
[1] = "Deploying NS at VIM."
2316 task_ro
= asyncio
.ensure_future(
2317 self
.instantiate_RO(
2318 logging_text
=logging_text
,
2322 db_nslcmop
=db_nslcmop
,
2325 n2vc_key_list
=n2vc_key_list
,
2329 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2330 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2332 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2333 stage
[1] = "Deploying Execution Environments."
2334 self
.logger
.debug(logging_text
+ stage
[1])
2336 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2337 for vnf_profile
in get_vnf_profiles(nsd
):
2338 vnfd_id
= vnf_profile
["vnfd-id"]
2339 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2340 member_vnf_index
= str(vnf_profile
["id"])
2341 db_vnfr
= db_vnfrs
[member_vnf_index
]
2342 base_folder
= vnfd
["_admin"]["storage"]
2348 # Get additional parameters
2349 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2350 if db_vnfr
.get("additionalParamsForVnf"):
2351 deploy_params
.update(
2352 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2355 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2356 if descriptor_config
:
2358 logging_text
=logging_text
2359 + "member_vnf_index={} ".format(member_vnf_index
),
2362 nslcmop_id
=nslcmop_id
,
2368 member_vnf_index
=member_vnf_index
,
2369 vdu_index
=vdu_index
,
2371 deploy_params
=deploy_params
,
2372 descriptor_config
=descriptor_config
,
2373 base_folder
=base_folder
,
2374 task_instantiation_info
=tasks_dict_info
,
2378 # Deploy charms for each VDU that supports one.
2379 for vdud
in get_vdu_list(vnfd
):
2381 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2382 vdur
= find_in_list(
2383 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2386 if vdur
.get("additionalParams"):
2387 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2389 deploy_params_vdu
= deploy_params
2390 deploy_params_vdu
["OSM"] = get_osm_params(
2391 db_vnfr
, vdu_id
, vdu_count_index
=0
2393 vdud_count
= get_vdu_profile(vnfd
, vdu_id
).get(
2394 "max-number-of-instances", 1
2397 self
.logger
.debug("VDUD > {}".format(vdud
))
2399 "Descriptor config > {}".format(descriptor_config
)
2401 if descriptor_config
:
2404 for vdu_index
in range(vdud_count
):
2405 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2407 logging_text
=logging_text
2408 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2409 member_vnf_index
, vdu_id
, vdu_index
2413 nslcmop_id
=nslcmop_id
,
2419 member_vnf_index
=member_vnf_index
,
2420 vdu_index
=vdu_index
,
2422 deploy_params
=deploy_params_vdu
,
2423 descriptor_config
=descriptor_config
,
2424 base_folder
=base_folder
,
2425 task_instantiation_info
=tasks_dict_info
,
2428 for kdud
in get_kdu_list(vnfd
):
2429 kdu_name
= kdud
["name"]
2430 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2431 if descriptor_config
:
2436 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2438 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2439 if kdur
.get("additionalParams"):
2440 deploy_params_kdu
= parse_yaml_strings(
2441 kdur
["additionalParams"]
2445 logging_text
=logging_text
,
2448 nslcmop_id
=nslcmop_id
,
2454 member_vnf_index
=member_vnf_index
,
2455 vdu_index
=vdu_index
,
2457 deploy_params
=deploy_params_kdu
,
2458 descriptor_config
=descriptor_config
,
2459 base_folder
=base_folder
,
2460 task_instantiation_info
=tasks_dict_info
,
2464 # Check if this NS has a charm configuration
2465 descriptor_config
= nsd
.get("ns-configuration")
2466 if descriptor_config
and descriptor_config
.get("juju"):
2469 member_vnf_index
= None
2475 # Get additional parameters
2476 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2477 if db_nsr
.get("additionalParamsForNs"):
2478 deploy_params
.update(
2479 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2481 base_folder
= nsd
["_admin"]["storage"]
2483 logging_text
=logging_text
,
2486 nslcmop_id
=nslcmop_id
,
2492 member_vnf_index
=member_vnf_index
,
2493 vdu_index
=vdu_index
,
2495 deploy_params
=deploy_params
,
2496 descriptor_config
=descriptor_config
,
2497 base_folder
=base_folder
,
2498 task_instantiation_info
=tasks_dict_info
,
2502 # rest of staff will be done at finally
2505 ROclient
.ROClientException
,
2511 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2514 except asyncio
.CancelledError
:
2516 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2518 exc
= "Operation was cancelled"
2519 except Exception as e
:
2520 exc
= traceback
.format_exc()
2521 self
.logger
.critical(
2522 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2527 error_list
.append(str(exc
))
2529 # wait for pending tasks
2531 stage
[1] = "Waiting for instantiate pending tasks."
2532 self
.logger
.debug(logging_text
+ stage
[1])
2533 error_list
+= await self
._wait
_for
_tasks
(
2541 stage
[1] = stage
[2] = ""
2542 except asyncio
.CancelledError
:
2543 error_list
.append("Cancelled")
2544 # TODO cancel all tasks
2545 except Exception as exc
:
2546 error_list
.append(str(exc
))
2548 # update operation-status
2549 db_nsr_update
["operational-status"] = "running"
2550 # let's begin with VCA 'configured' status (later we can change it)
2551 db_nsr_update
["config-status"] = "configured"
2552 for task
, task_name
in tasks_dict_info
.items():
2553 if not task
.done() or task
.cancelled() or task
.exception():
2554 if task_name
.startswith(self
.task_name_deploy_vca
):
2555 # A N2VC task is pending
2556 db_nsr_update
["config-status"] = "failed"
2558 # RO or KDU task is pending
2559 db_nsr_update
["operational-status"] = "failed"
2561 # update status at database
2563 error_detail
= ". ".join(error_list
)
2564 self
.logger
.error(logging_text
+ error_detail
)
2565 error_description_nslcmop
= "{} Detail: {}".format(
2566 stage
[0], error_detail
2568 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2569 nslcmop_id
, stage
[0]
2572 db_nsr_update
["detailed-status"] = (
2573 error_description_nsr
+ " Detail: " + error_detail
2575 db_nslcmop_update
["detailed-status"] = error_detail
2576 nslcmop_operation_state
= "FAILED"
2580 error_description_nsr
= error_description_nslcmop
= None
2582 db_nsr_update
["detailed-status"] = "Done"
2583 db_nslcmop_update
["detailed-status"] = "Done"
2584 nslcmop_operation_state
= "COMPLETED"
2587 self
._write
_ns
_status
(
2590 current_operation
="IDLE",
2591 current_operation_id
=None,
2592 error_description
=error_description_nsr
,
2593 error_detail
=error_detail
,
2594 other_update
=db_nsr_update
,
2596 self
._write
_op
_status
(
2599 error_message
=error_description_nslcmop
,
2600 operation_state
=nslcmop_operation_state
,
2601 other_update
=db_nslcmop_update
,
2604 if nslcmop_operation_state
:
2606 await self
.msg
.aiowrite(
2611 "nslcmop_id": nslcmop_id
,
2612 "operationState": nslcmop_operation_state
,
2616 except Exception as e
:
2618 logging_text
+ "kafka_write notification Exception {}".format(e
)
2621 self
.logger
.debug(logging_text
+ "Exit")
2622 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2624 async def _add_vca_relations(
2629 timeout
: int = 3600,
2630 vca_type
: str = None,
2635 # 1. find all relations for this VCA
2636 # 2. wait for other peers related
2640 vca_type
= vca_type
or "lxc_proxy_charm"
2642 # STEP 1: find all relations for this VCA
2645 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2646 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2649 my_vca
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))[vca_index
]
2651 # read all ns-configuration relations
2652 ns_relations
= list()
2653 db_ns_relations
= deep_get(nsd
, ("ns-configuration", "relation"))
2655 for r
in db_ns_relations
:
2656 # check if this VCA is in the relation
2657 if my_vca
.get("member-vnf-index") in (
2658 r
.get("entities")[0].get("id"),
2659 r
.get("entities")[1].get("id"),
2661 ns_relations
.append(r
)
2663 # read all vnf-configuration relations
2664 vnf_relations
= list()
2665 db_vnfd_list
= db_nsr
.get("vnfd-id")
2667 for vnfd
in db_vnfd_list
:
2668 db_vnf_relations
= None
2669 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2670 db_vnf_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
2671 if db_vnf_configuration
:
2672 db_vnf_relations
= db_vnf_configuration
.get("relation", [])
2673 if db_vnf_relations
:
2674 for r
in db_vnf_relations
:
2675 # check if this VCA is in the relation
2676 if my_vca
.get("vdu_id") in (
2677 r
.get("entities")[0].get("id"),
2678 r
.get("entities")[1].get("id"),
2680 vnf_relations
.append(r
)
2682 # if no relations, terminate
2683 if not ns_relations
and not vnf_relations
:
2684 self
.logger
.debug(logging_text
+ " No relations")
2689 + " adding relations\n {}\n {}".format(
2690 ns_relations
, vnf_relations
2699 if now
- start
>= timeout
:
2700 self
.logger
.error(logging_text
+ " : timeout adding relations")
2703 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2704 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2706 # for each defined NS relation, find the VCA's related
2707 for r
in ns_relations
.copy():
2708 from_vca_ee_id
= None
2710 from_vca_endpoint
= None
2711 to_vca_endpoint
= None
2712 vca_list
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))
2713 for vca
in vca_list
:
2714 if vca
.get("member-vnf-index") == r
.get("entities")[0].get(
2716 ) and vca
.get("config_sw_installed"):
2717 from_vca_ee_id
= vca
.get("ee_id")
2718 from_vca_endpoint
= r
.get("entities")[0].get("endpoint")
2719 if vca
.get("member-vnf-index") == r
.get("entities")[1].get(
2721 ) and vca
.get("config_sw_installed"):
2722 to_vca_ee_id
= vca
.get("ee_id")
2723 to_vca_endpoint
= r
.get("entities")[1].get("endpoint")
2724 if from_vca_ee_id
and to_vca_ee_id
:
2726 await self
.vca_map
[vca_type
].add_relation(
2727 ee_id_1
=from_vca_ee_id
,
2728 ee_id_2
=to_vca_ee_id
,
2729 endpoint_1
=from_vca_endpoint
,
2730 endpoint_2
=to_vca_endpoint
,
2733 # remove entry from relations list
2734 ns_relations
.remove(r
)
2736 # check failed peers
2738 vca_status_list
= db_nsr
.get("configurationStatus")
2740 for i
in range(len(vca_list
)):
2742 vca_status
= vca_status_list
[i
]
2743 if vca
.get("member-vnf-index") == r
.get("entities")[
2746 if vca_status
.get("status") == "BROKEN":
2747 # peer broken: remove relation from list
2748 ns_relations
.remove(r
)
2749 if vca
.get("member-vnf-index") == r
.get("entities")[
2752 if vca_status
.get("status") == "BROKEN":
2753 # peer broken: remove relation from list
2754 ns_relations
.remove(r
)
2759 # for each defined VNF relation, find the VCA's related
2760 for r
in vnf_relations
.copy():
2761 from_vca_ee_id
= None
2763 from_vca_endpoint
= None
2764 to_vca_endpoint
= None
2765 vca_list
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))
2766 for vca
in vca_list
:
2767 key_to_check
= "vdu_id"
2768 if vca
.get("vdu_id") is None:
2769 key_to_check
= "vnfd_id"
2770 if vca
.get(key_to_check
) == 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(key_to_check
) == 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 vnf_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("vdu_id") == r
.get("entities")[0].get(
2802 if vca_status
.get("status") == "BROKEN":
2803 # peer broken: remove relation from list
2804 vnf_relations
.remove(r
)
2805 if vca
.get("vdu_id") == r
.get("entities")[1].get(
2808 if vca_status
.get("status") == "BROKEN":
2809 # peer broken: remove relation from list
2810 vnf_relations
.remove(r
)
2816 await asyncio
.sleep(5.0)
2818 if not ns_relations
and not vnf_relations
:
2819 self
.logger
.debug("Relations added")
2824 except Exception as e
:
2825 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
2828 async def _install_kdu(
2836 k8s_instance_info
: dict,
2837 k8params
: dict = None,
2843 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
2846 "collection": "nsrs",
2847 "filter": {"_id": nsr_id
},
2848 "path": nsr_db_path
,
2851 if k8s_instance_info
.get("kdu-deployment-name"):
2852 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
2854 kdu_instance
= self
.k8scluster_map
[
2856 ].generate_kdu_instance_name(
2857 db_dict
=db_dict_install
,
2858 kdu_model
=k8s_instance_info
["kdu-model"],
2859 kdu_name
=k8s_instance_info
["kdu-name"],
2862 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
2864 await self
.k8scluster_map
[k8sclustertype
].install(
2865 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2866 kdu_model
=k8s_instance_info
["kdu-model"],
2869 db_dict
=db_dict_install
,
2871 kdu_name
=k8s_instance_info
["kdu-name"],
2872 namespace
=k8s_instance_info
["namespace"],
2873 kdu_instance
=kdu_instance
,
2877 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
2880 # Obtain services to obtain management service ip
2881 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
2882 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2883 kdu_instance
=kdu_instance
,
2884 namespace
=k8s_instance_info
["namespace"],
2887 # Obtain management service info (if exists)
2888 vnfr_update_dict
= {}
2889 kdu_config
= get_configuration(vnfd
, kdud
["name"])
2891 target_ee_list
= kdu_config
.get("execution-environment-list", [])
2896 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
2899 for service
in kdud
.get("service", [])
2900 if service
.get("mgmt-service")
2902 for mgmt_service
in mgmt_services
:
2903 for service
in services
:
2904 if service
["name"].startswith(mgmt_service
["name"]):
2905 # Mgmt service found, Obtain service ip
2906 ip
= service
.get("external_ip", service
.get("cluster_ip"))
2907 if isinstance(ip
, list) and len(ip
) == 1:
2911 "kdur.{}.ip-address".format(kdu_index
)
2914 # Check if must update also mgmt ip at the vnf
2915 service_external_cp
= mgmt_service
.get(
2916 "external-connection-point-ref"
2918 if service_external_cp
:
2920 deep_get(vnfd
, ("mgmt-interface", "cp"))
2921 == service_external_cp
2923 vnfr_update_dict
["ip-address"] = ip
2928 "external-connection-point-ref", ""
2930 == service_external_cp
,
2933 "kdur.{}.ip-address".format(kdu_index
)
2938 "Mgmt service name: {} not found".format(
2939 mgmt_service
["name"]
2943 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
2944 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
2946 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
2949 and kdu_config
.get("initial-config-primitive")
2950 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
2952 initial_config_primitive_list
= kdu_config
.get(
2953 "initial-config-primitive"
2955 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
2957 for initial_config_primitive
in initial_config_primitive_list
:
2958 primitive_params_
= self
._map
_primitive
_params
(
2959 initial_config_primitive
, {}, {}
2962 await asyncio
.wait_for(
2963 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
2964 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2965 kdu_instance
=kdu_instance
,
2966 primitive_name
=initial_config_primitive
["name"],
2967 params
=primitive_params_
,
2968 db_dict
=db_dict_install
,
2974 except Exception as e
:
2975 # Prepare update db with error and raise exception
2978 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
2982 vnfr_data
.get("_id"),
2983 {"kdur.{}.status".format(kdu_index
): "ERROR"},
2986 # ignore to keep original exception
2988 # reraise original error
2993 async def deploy_kdus(
3000 task_instantiation_info
,
3002 # Launch kdus if present in the descriptor
3004 k8scluster_id_2_uuic
= {
3005 "helm-chart-v3": {},
3010 async def _get_cluster_id(cluster_id
, cluster_type
):
3011 nonlocal k8scluster_id_2_uuic
3012 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3013 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3015 # check if K8scluster is creating and wait look if previous tasks in process
3016 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3017 "k8scluster", cluster_id
3020 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3021 task_name
, cluster_id
3023 self
.logger
.debug(logging_text
+ text
)
3024 await asyncio
.wait(task_dependency
, timeout
=3600)
3026 db_k8scluster
= self
.db
.get_one(
3027 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3029 if not db_k8scluster
:
3030 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3032 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3034 if cluster_type
== "helm-chart-v3":
3036 # backward compatibility for existing clusters that have not been initialized for helm v3
3037 k8s_credentials
= yaml
.safe_dump(
3038 db_k8scluster
.get("credentials")
3040 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3041 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3043 db_k8scluster_update
= {}
3044 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3045 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3046 db_k8scluster_update
[
3047 "_admin.helm-chart-v3.created"
3049 db_k8scluster_update
[
3050 "_admin.helm-chart-v3.operationalState"
3053 "k8sclusters", cluster_id
, db_k8scluster_update
3055 except Exception as e
:
3058 + "error initializing helm-v3 cluster: {}".format(str(e
))
3061 "K8s cluster '{}' has not been initialized for '{}'".format(
3062 cluster_id
, cluster_type
3067 "K8s cluster '{}' has not been initialized for '{}'".format(
3068 cluster_id
, cluster_type
3071 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3074 logging_text
+= "Deploy kdus: "
3077 db_nsr_update
= {"_admin.deployed.K8s": []}
3078 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3081 updated_cluster_list
= []
3082 updated_v3_cluster_list
= []
3084 for vnfr_data
in db_vnfrs
.values():
3085 vca_id
= self
.get_vca_id(vnfr_data
, {})
3086 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3087 # Step 0: Prepare and set parameters
3088 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3089 vnfd_id
= vnfr_data
.get("vnfd-id")
3090 vnfd_with_id
= find_in_list(
3091 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3095 for kdud
in vnfd_with_id
["kdu"]
3096 if kdud
["name"] == kdur
["kdu-name"]
3098 namespace
= kdur
.get("k8s-namespace")
3099 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3100 if kdur
.get("helm-chart"):
3101 kdumodel
= kdur
["helm-chart"]
3102 # Default version: helm3, if helm-version is v2 assign v2
3103 k8sclustertype
= "helm-chart-v3"
3104 self
.logger
.debug("kdur: {}".format(kdur
))
3106 kdur
.get("helm-version")
3107 and kdur
.get("helm-version") == "v2"
3109 k8sclustertype
= "helm-chart"
3110 elif kdur
.get("juju-bundle"):
3111 kdumodel
= kdur
["juju-bundle"]
3112 k8sclustertype
= "juju-bundle"
3115 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3116 "juju-bundle. Maybe an old NBI version is running".format(
3117 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3120 # check if kdumodel is a file and exists
3122 vnfd_with_id
= find_in_list(
3123 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3125 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3126 if storage
and storage
.get(
3128 ): # may be not present if vnfd has not artifacts
3129 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3130 filename
= "{}/{}/{}s/{}".format(
3136 if self
.fs
.file_exists(
3137 filename
, mode
="file"
3138 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3139 kdumodel
= self
.fs
.path
+ filename
3140 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3142 except Exception: # it is not a file
3145 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3146 step
= "Synchronize repos for k8s cluster '{}'".format(
3149 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3153 k8sclustertype
== "helm-chart"
3154 and cluster_uuid
not in updated_cluster_list
3156 k8sclustertype
== "helm-chart-v3"
3157 and cluster_uuid
not in updated_v3_cluster_list
3159 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3160 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3161 cluster_uuid
=cluster_uuid
3164 if del_repo_list
or added_repo_dict
:
3165 if k8sclustertype
== "helm-chart":
3167 "_admin.helm_charts_added." + item
: None
3168 for item
in del_repo_list
3171 "_admin.helm_charts_added." + item
: name
3172 for item
, name
in added_repo_dict
.items()
3174 updated_cluster_list
.append(cluster_uuid
)
3175 elif k8sclustertype
== "helm-chart-v3":
3177 "_admin.helm_charts_v3_added." + item
: None
3178 for item
in del_repo_list
3181 "_admin.helm_charts_v3_added." + item
: name
3182 for item
, name
in added_repo_dict
.items()
3184 updated_v3_cluster_list
.append(cluster_uuid
)
3186 logging_text
+ "repos synchronized on k8s cluster "
3187 "'{}' to_delete: {}, to_add: {}".format(
3188 k8s_cluster_id
, del_repo_list
, added_repo_dict
3193 {"_id": k8s_cluster_id
},
3199 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3200 vnfr_data
["member-vnf-index-ref"],
3204 k8s_instance_info
= {
3205 "kdu-instance": None,
3206 "k8scluster-uuid": cluster_uuid
,
3207 "k8scluster-type": k8sclustertype
,
3208 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3209 "kdu-name": kdur
["kdu-name"],
3210 "kdu-model": kdumodel
,
3211 "namespace": namespace
,
3212 "kdu-deployment-name": kdu_deployment_name
,
3214 db_path
= "_admin.deployed.K8s.{}".format(index
)
3215 db_nsr_update
[db_path
] = k8s_instance_info
3216 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3217 vnfd_with_id
= find_in_list(
3218 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3220 task
= asyncio
.ensure_future(
3229 k8params
=desc_params
,
3234 self
.lcm_tasks
.register(
3238 "instantiate_KDU-{}".format(index
),
3241 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3247 except (LcmException
, asyncio
.CancelledError
):
3249 except Exception as e
:
3250 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3251 if isinstance(e
, (N2VCException
, DbException
)):
3252 self
.logger
.error(logging_text
+ msg
)
3254 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3255 raise LcmException(msg
)
3258 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3277 task_instantiation_info
,
3280 # launch instantiate_N2VC in a asyncio task and register task object
3281 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3282 # if not found, create one entry and update database
3283 # fill db_nsr._admin.deployed.VCA.<index>
3286 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3288 if "execution-environment-list" in descriptor_config
:
3289 ee_list
= descriptor_config
.get("execution-environment-list", [])
3290 elif "juju" in descriptor_config
:
3291 ee_list
= [descriptor_config
] # ns charms
3292 else: # other types as script are not supported
3295 for ee_item
in ee_list
:
3298 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3299 ee_item
.get("juju"), ee_item
.get("helm-chart")
3302 ee_descriptor_id
= ee_item
.get("id")
3303 if ee_item
.get("juju"):
3304 vca_name
= ee_item
["juju"].get("charm")
3307 if ee_item
["juju"].get("charm") is not None
3310 if ee_item
["juju"].get("cloud") == "k8s":
3311 vca_type
= "k8s_proxy_charm"
3312 elif ee_item
["juju"].get("proxy") is False:
3313 vca_type
= "native_charm"
3314 elif ee_item
.get("helm-chart"):
3315 vca_name
= ee_item
["helm-chart"]
3316 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3319 vca_type
= "helm-v3"
3322 logging_text
+ "skipping non juju neither charm configuration"
3327 for vca_index
, vca_deployed
in enumerate(
3328 db_nsr
["_admin"]["deployed"]["VCA"]
3330 if not vca_deployed
:
3333 vca_deployed
.get("member-vnf-index") == member_vnf_index
3334 and vca_deployed
.get("vdu_id") == vdu_id
3335 and vca_deployed
.get("kdu_name") == kdu_name
3336 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3337 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3341 # not found, create one.
3343 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3346 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3348 target
+= "/kdu/{}".format(kdu_name
)
3350 "target_element": target
,
3351 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3352 "member-vnf-index": member_vnf_index
,
3354 "kdu_name": kdu_name
,
3355 "vdu_count_index": vdu_index
,
3356 "operational-status": "init", # TODO revise
3357 "detailed-status": "", # TODO revise
3358 "step": "initial-deploy", # TODO revise
3360 "vdu_name": vdu_name
,
3362 "ee_descriptor_id": ee_descriptor_id
,
3366 # create VCA and configurationStatus in db
3368 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3369 "configurationStatus.{}".format(vca_index
): dict(),
3371 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3373 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3375 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3376 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3377 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3380 task_n2vc
= asyncio
.ensure_future(
3381 self
.instantiate_N2VC(
3382 logging_text
=logging_text
,
3383 vca_index
=vca_index
,
3389 vdu_index
=vdu_index
,
3390 deploy_params
=deploy_params
,
3391 config_descriptor
=descriptor_config
,
3392 base_folder
=base_folder
,
3393 nslcmop_id
=nslcmop_id
,
3397 ee_config_descriptor
=ee_item
,
3400 self
.lcm_tasks
.register(
3404 "instantiate_N2VC-{}".format(vca_index
),
3407 task_instantiation_info
[
3409 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3410 member_vnf_index
or "", vdu_id
or ""
3414 def _create_nslcmop(nsr_id
, operation
, params
):
3416 Creates a ns-lcm-opp content to be stored at database.
3417 :param nsr_id: internal id of the instance
3418 :param operation: instantiate, terminate, scale, action, ...
3419 :param params: user parameters for the operation
3420 :return: dictionary following SOL005 format
3422 # Raise exception if invalid arguments
3423 if not (nsr_id
and operation
and params
):
3425 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3432 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3433 "operationState": "PROCESSING",
3434 "statusEnteredTime": now
,
3435 "nsInstanceId": nsr_id
,
3436 "lcmOperationType": operation
,
3438 "isAutomaticInvocation": False,
3439 "operationParams": params
,
3440 "isCancelPending": False,
3442 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3443 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3448 def _format_additional_params(self
, params
):
3449 params
= params
or {}
3450 for key
, value
in params
.items():
3451 if str(value
).startswith("!!yaml "):
3452 params
[key
] = yaml
.safe_load(value
[7:])
3455 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3456 primitive
= seq
.get("name")
3457 primitive_params
= {}
3459 "member_vnf_index": vnf_index
,
3460 "primitive": primitive
,
3461 "primitive_params": primitive_params
,
3464 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3468 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3469 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3470 if op
.get("operationState") == "COMPLETED":
3471 # b. Skip sub-operation
3472 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3473 return self
.SUBOPERATION_STATUS_SKIP
3475 # c. retry executing sub-operation
3476 # The sub-operation exists, and operationState != 'COMPLETED'
3477 # Update operationState = 'PROCESSING' to indicate a retry.
3478 operationState
= "PROCESSING"
3479 detailed_status
= "In progress"
3480 self
._update
_suboperation
_status
(
3481 db_nslcmop
, op_index
, operationState
, detailed_status
3483 # Return the sub-operation index
3484 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3485 # with arguments extracted from the sub-operation
3488 # Find a sub-operation where all keys in a matching dictionary must match
3489 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3490 def _find_suboperation(self
, db_nslcmop
, match
):
3491 if db_nslcmop
and match
:
3492 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3493 for i
, op
in enumerate(op_list
):
3494 if all(op
.get(k
) == match
[k
] for k
in match
):
3496 return self
.SUBOPERATION_STATUS_NOT_FOUND
3498 # Update status for a sub-operation given its index
3499 def _update_suboperation_status(
3500 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3502 # Update DB for HA tasks
3503 q_filter
= {"_id": db_nslcmop
["_id"]}
3505 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3506 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3509 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3512 # Add sub-operation, return the index of the added sub-operation
3513 # Optionally, set operationState, detailed-status, and operationType
3514 # Status and type are currently set for 'scale' sub-operations:
3515 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3516 # 'detailed-status' : status message
3517 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3518 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3519 def _add_suboperation(
3527 mapped_primitive_params
,
3528 operationState
=None,
3529 detailed_status
=None,
3532 RO_scaling_info
=None,
3535 return self
.SUBOPERATION_STATUS_NOT_FOUND
3536 # Get the "_admin.operations" list, if it exists
3537 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3538 op_list
= db_nslcmop_admin
.get("operations")
3539 # Create or append to the "_admin.operations" list
3541 "member_vnf_index": vnf_index
,
3543 "vdu_count_index": vdu_count_index
,
3544 "primitive": primitive
,
3545 "primitive_params": mapped_primitive_params
,
3548 new_op
["operationState"] = operationState
3550 new_op
["detailed-status"] = detailed_status
3552 new_op
["lcmOperationType"] = operationType
3554 new_op
["RO_nsr_id"] = RO_nsr_id
3556 new_op
["RO_scaling_info"] = RO_scaling_info
3558 # No existing operations, create key 'operations' with current operation as first list element
3559 db_nslcmop_admin
.update({"operations": [new_op
]})
3560 op_list
= db_nslcmop_admin
.get("operations")
3562 # Existing operations, append operation to list
3563 op_list
.append(new_op
)
3565 db_nslcmop_update
= {"_admin.operations": op_list
}
3566 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3567 op_index
= len(op_list
) - 1
3570 # Helper methods for scale() sub-operations
3572 # pre-scale/post-scale:
3573 # Check for 3 different cases:
3574 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3575 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3576 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3577 def _check_or_add_scale_suboperation(
3581 vnf_config_primitive
,
3585 RO_scaling_info
=None,
3587 # Find this sub-operation
3588 if RO_nsr_id
and RO_scaling_info
:
3589 operationType
= "SCALE-RO"
3591 "member_vnf_index": vnf_index
,
3592 "RO_nsr_id": RO_nsr_id
,
3593 "RO_scaling_info": RO_scaling_info
,
3597 "member_vnf_index": vnf_index
,
3598 "primitive": vnf_config_primitive
,
3599 "primitive_params": primitive_params
,
3600 "lcmOperationType": operationType
,
3602 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3603 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3604 # a. New sub-operation
3605 # The sub-operation does not exist, add it.
3606 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3607 # The following parameters are set to None for all kind of scaling:
3609 vdu_count_index
= None
3611 if RO_nsr_id
and RO_scaling_info
:
3612 vnf_config_primitive
= None
3613 primitive_params
= None
3616 RO_scaling_info
= None
3617 # Initial status for sub-operation
3618 operationState
= "PROCESSING"
3619 detailed_status
= "In progress"
3620 # Add sub-operation for pre/post-scaling (zero or more operations)
3621 self
._add
_suboperation
(
3627 vnf_config_primitive
,
3635 return self
.SUBOPERATION_STATUS_NEW
3637 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3638 # or op_index (operationState != 'COMPLETED')
3639 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3641 # Function to return execution_environment id
3643 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3644 # TODO vdu_index_count
3645 for vca
in vca_deployed_list
:
3646 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3649 async def destroy_N2VC(
3657 exec_primitives
=True,
3662 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3663 :param logging_text:
3665 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3666 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3667 :param vca_index: index in the database _admin.deployed.VCA
3668 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3669 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3670 not executed properly
3671 :param scaling_in: True destroys the application, False destroys the model
3672 :return: None or exception
3677 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3678 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3682 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3684 # execute terminate_primitives
3686 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3687 config_descriptor
.get("terminate-config-primitive"),
3688 vca_deployed
.get("ee_descriptor_id"),
3690 vdu_id
= vca_deployed
.get("vdu_id")
3691 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3692 vdu_name
= vca_deployed
.get("vdu_name")
3693 vnf_index
= vca_deployed
.get("member-vnf-index")
3694 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3695 for seq
in terminate_primitives
:
3696 # For each sequence in list, get primitive and call _ns_execute_primitive()
3697 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3698 vnf_index
, seq
.get("name")
3700 self
.logger
.debug(logging_text
+ step
)
3701 # Create the primitive for each sequence, i.e. "primitive": "touch"
3702 primitive
= seq
.get("name")
3703 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3708 self
._add
_suboperation
(
3715 mapped_primitive_params
,
3717 # Sub-operations: Call _ns_execute_primitive() instead of action()
3719 result
, result_detail
= await self
._ns
_execute
_primitive
(
3720 vca_deployed
["ee_id"],
3722 mapped_primitive_params
,
3726 except LcmException
:
3727 # this happens when VCA is not deployed. In this case it is not needed to terminate
3729 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3730 if result
not in result_ok
:
3732 "terminate_primitive {} for vnf_member_index={} fails with "
3733 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3735 # set that this VCA do not need terminated
3736 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3740 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3743 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3744 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3747 await self
.vca_map
[vca_type
].delete_execution_environment(
3748 vca_deployed
["ee_id"],
3749 scaling_in
=scaling_in
,
3753 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3754 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3755 namespace
= "." + db_nsr
["_id"]
3757 await self
.n2vc
.delete_namespace(
3758 namespace
=namespace
,
3759 total_timeout
=self
.timeout_charm_delete
,
3762 except N2VCNotFound
: # already deleted. Skip
3764 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3766 async def _terminate_RO(
3767 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
3770 Terminates a deployment from RO
3771 :param logging_text:
3772 :param nsr_deployed: db_nsr._admin.deployed
3775 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3776 this method will update only the index 2, but it will write on database the concatenated content of the list
3781 ro_nsr_id
= ro_delete_action
= None
3782 if nsr_deployed
and nsr_deployed
.get("RO"):
3783 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3784 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3787 stage
[2] = "Deleting ns from VIM."
3788 db_nsr_update
["detailed-status"] = " ".join(stage
)
3789 self
._write
_op
_status
(nslcmop_id
, stage
)
3790 self
.logger
.debug(logging_text
+ stage
[2])
3791 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3792 self
._write
_op
_status
(nslcmop_id
, stage
)
3793 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3794 ro_delete_action
= desc
["action_id"]
3796 "_admin.deployed.RO.nsr_delete_action_id"
3797 ] = ro_delete_action
3798 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3799 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3800 if ro_delete_action
:
3801 # wait until NS is deleted from VIM
3802 stage
[2] = "Waiting ns deleted from VIM."
3803 detailed_status_old
= None
3807 + " RO_id={} ro_delete_action={}".format(
3808 ro_nsr_id
, ro_delete_action
3811 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3812 self
._write
_op
_status
(nslcmop_id
, stage
)
3814 delete_timeout
= 20 * 60 # 20 minutes
3815 while delete_timeout
> 0:
3816 desc
= await self
.RO
.show(
3818 item_id_name
=ro_nsr_id
,
3819 extra_item
="action",
3820 extra_item_id
=ro_delete_action
,
3824 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3826 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3827 if ns_status
== "ERROR":
3828 raise ROclient
.ROClientException(ns_status_info
)
3829 elif ns_status
== "BUILD":
3830 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3831 elif ns_status
== "ACTIVE":
3832 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3833 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3838 ), "ROclient.check_action_status returns unknown {}".format(
3841 if stage
[2] != detailed_status_old
:
3842 detailed_status_old
= stage
[2]
3843 db_nsr_update
["detailed-status"] = " ".join(stage
)
3844 self
._write
_op
_status
(nslcmop_id
, stage
)
3845 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3846 await asyncio
.sleep(5, loop
=self
.loop
)
3848 else: # delete_timeout <= 0:
3849 raise ROclient
.ROClientException(
3850 "Timeout waiting ns deleted from VIM"
3853 except Exception as e
:
3854 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3856 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3858 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3859 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3860 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3862 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
3865 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
3867 failed_detail
.append("delete conflict: {}".format(e
))
3870 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
3873 failed_detail
.append("delete error: {}".format(e
))
3875 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
3879 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
3880 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
3882 stage
[2] = "Deleting nsd from RO."
3883 db_nsr_update
["detailed-status"] = " ".join(stage
)
3884 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3885 self
._write
_op
_status
(nslcmop_id
, stage
)
3886 await self
.RO
.delete("nsd", ro_nsd_id
)
3888 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
3890 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3891 except Exception as e
:
3893 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3895 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3897 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
3900 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
3902 failed_detail
.append(
3903 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
3905 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3907 failed_detail
.append(
3908 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
3910 self
.logger
.error(logging_text
+ failed_detail
[-1])
3912 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
3913 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
3914 if not vnf_deployed
or not vnf_deployed
["id"]:
3917 ro_vnfd_id
= vnf_deployed
["id"]
3920 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
3921 vnf_deployed
["member-vnf-index"], ro_vnfd_id
3923 db_nsr_update
["detailed-status"] = " ".join(stage
)
3924 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3925 self
._write
_op
_status
(nslcmop_id
, stage
)
3926 await self
.RO
.delete("vnfd", ro_vnfd_id
)
3928 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
3930 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3931 except Exception as e
:
3933 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3936 "_admin.deployed.RO.vnfd.{}.id".format(index
)
3940 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
3943 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
3945 failed_detail
.append(
3946 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
3948 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3950 failed_detail
.append(
3951 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
3953 self
.logger
.error(logging_text
+ failed_detail
[-1])
3956 stage
[2] = "Error deleting from VIM"
3958 stage
[2] = "Deleted from VIM"
3959 db_nsr_update
["detailed-status"] = " ".join(stage
)
3960 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3961 self
._write
_op
_status
(nslcmop_id
, stage
)
3964 raise LcmException("; ".join(failed_detail
))
3966 async def terminate(self
, nsr_id
, nslcmop_id
):
3967 # Try to lock HA task here
3968 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
3969 if not task_is_locked_by_me
:
3972 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
3973 self
.logger
.debug(logging_text
+ "Enter")
3974 timeout_ns_terminate
= self
.timeout_ns_terminate
3977 operation_params
= None
3979 error_list
= [] # annotates all failed error messages
3980 db_nslcmop_update
= {}
3981 autoremove
= False # autoremove after terminated
3982 tasks_dict_info
= {}
3985 "Stage 1/3: Preparing task.",
3986 "Waiting for previous operations to terminate.",
3989 # ^ contains [stage, step, VIM-status]
3991 # wait for any previous tasks in process
3992 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
3994 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
3995 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3996 operation_params
= db_nslcmop
.get("operationParams") or {}
3997 if operation_params
.get("timeout_ns_terminate"):
3998 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
3999 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4000 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4002 db_nsr_update
["operational-status"] = "terminating"
4003 db_nsr_update
["config-status"] = "terminating"
4004 self
._write
_ns
_status
(
4006 ns_state
="TERMINATING",
4007 current_operation
="TERMINATING",
4008 current_operation_id
=nslcmop_id
,
4009 other_update
=db_nsr_update
,
4011 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4012 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4013 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4016 stage
[1] = "Getting vnf descriptors from db."
4017 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4019 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4021 db_vnfds_from_id
= {}
4022 db_vnfds_from_member_index
= {}
4024 for vnfr
in db_vnfrs_list
:
4025 vnfd_id
= vnfr
["vnfd-id"]
4026 if vnfd_id
not in db_vnfds_from_id
:
4027 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4028 db_vnfds_from_id
[vnfd_id
] = vnfd
4029 db_vnfds_from_member_index
[
4030 vnfr
["member-vnf-index-ref"]
4031 ] = db_vnfds_from_id
[vnfd_id
]
4033 # Destroy individual execution environments when there are terminating primitives.
4034 # Rest of EE will be deleted at once
4035 # TODO - check before calling _destroy_N2VC
4036 # if not operation_params.get("skip_terminate_primitives"):#
4037 # or not vca.get("needed_terminate"):
4038 stage
[0] = "Stage 2/3 execute terminating primitives."
4039 self
.logger
.debug(logging_text
+ stage
[0])
4040 stage
[1] = "Looking execution environment that needs terminate."
4041 self
.logger
.debug(logging_text
+ stage
[1])
4043 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4044 config_descriptor
= None
4045 vca_member_vnf_index
= vca
.get("member-vnf-index")
4046 vca_id
= self
.get_vca_id(
4047 db_vnfrs_dict
.get(vca_member_vnf_index
)
4048 if vca_member_vnf_index
4052 if not vca
or not vca
.get("ee_id"):
4054 if not vca
.get("member-vnf-index"):
4056 config_descriptor
= db_nsr
.get("ns-configuration")
4057 elif vca
.get("vdu_id"):
4058 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4059 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4060 elif vca
.get("kdu_name"):
4061 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4062 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4064 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4065 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4066 vca_type
= vca
.get("type")
4067 exec_terminate_primitives
= not operation_params
.get(
4068 "skip_terminate_primitives"
4069 ) and vca
.get("needed_terminate")
4070 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4071 # pending native charms
4073 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4075 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4076 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4077 task
= asyncio
.ensure_future(
4085 exec_terminate_primitives
,
4089 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4091 # wait for pending tasks of terminate primitives
4095 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4097 error_list
= await self
._wait
_for
_tasks
(
4100 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4104 tasks_dict_info
.clear()
4106 return # raise LcmException("; ".join(error_list))
4108 # remove All execution environments at once
4109 stage
[0] = "Stage 3/3 delete all."
4111 if nsr_deployed
.get("VCA"):
4112 stage
[1] = "Deleting all execution environments."
4113 self
.logger
.debug(logging_text
+ stage
[1])
4114 vca_id
= self
.get_vca_id({}, db_nsr
)
4115 task_delete_ee
= asyncio
.ensure_future(
4117 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4118 timeout
=self
.timeout_charm_delete
,
4121 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4122 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4124 # Delete from k8scluster
4125 stage
[1] = "Deleting KDUs."
4126 self
.logger
.debug(logging_text
+ stage
[1])
4127 # print(nsr_deployed)
4128 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4129 if not kdu
or not kdu
.get("kdu-instance"):
4131 kdu_instance
= kdu
.get("kdu-instance")
4132 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4133 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4134 vca_id
= self
.get_vca_id({}, db_nsr
)
4135 task_delete_kdu_instance
= asyncio
.ensure_future(
4136 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4137 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4138 kdu_instance
=kdu_instance
,
4145 + "Unknown k8s deployment type {}".format(
4146 kdu
.get("k8scluster-type")
4151 task_delete_kdu_instance
4152 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4155 stage
[1] = "Deleting ns from VIM."
4157 task_delete_ro
= asyncio
.ensure_future(
4158 self
._terminate
_ng
_ro
(
4159 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4163 task_delete_ro
= asyncio
.ensure_future(
4165 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4168 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4170 # rest of staff will be done at finally
4173 ROclient
.ROClientException
,
4178 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4180 except asyncio
.CancelledError
:
4182 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4184 exc
= "Operation was cancelled"
4185 except Exception as e
:
4186 exc
= traceback
.format_exc()
4187 self
.logger
.critical(
4188 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4193 error_list
.append(str(exc
))
4195 # wait for pending tasks
4197 stage
[1] = "Waiting for terminate pending tasks."
4198 self
.logger
.debug(logging_text
+ stage
[1])
4199 error_list
+= await self
._wait
_for
_tasks
(
4202 timeout_ns_terminate
,
4206 stage
[1] = stage
[2] = ""
4207 except asyncio
.CancelledError
:
4208 error_list
.append("Cancelled")
4209 # TODO cancell all tasks
4210 except Exception as exc
:
4211 error_list
.append(str(exc
))
4212 # update status at database
4214 error_detail
= "; ".join(error_list
)
4215 # self.logger.error(logging_text + error_detail)
4216 error_description_nslcmop
= "{} Detail: {}".format(
4217 stage
[0], error_detail
4219 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4220 nslcmop_id
, stage
[0]
4223 db_nsr_update
["operational-status"] = "failed"
4224 db_nsr_update
["detailed-status"] = (
4225 error_description_nsr
+ " Detail: " + error_detail
4227 db_nslcmop_update
["detailed-status"] = error_detail
4228 nslcmop_operation_state
= "FAILED"
4232 error_description_nsr
= error_description_nslcmop
= None
4233 ns_state
= "NOT_INSTANTIATED"
4234 db_nsr_update
["operational-status"] = "terminated"
4235 db_nsr_update
["detailed-status"] = "Done"
4236 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4237 db_nslcmop_update
["detailed-status"] = "Done"
4238 nslcmop_operation_state
= "COMPLETED"
4241 self
._write
_ns
_status
(
4244 current_operation
="IDLE",
4245 current_operation_id
=None,
4246 error_description
=error_description_nsr
,
4247 error_detail
=error_detail
,
4248 other_update
=db_nsr_update
,
4250 self
._write
_op
_status
(
4253 error_message
=error_description_nslcmop
,
4254 operation_state
=nslcmop_operation_state
,
4255 other_update
=db_nslcmop_update
,
4257 if ns_state
== "NOT_INSTANTIATED":
4261 {"nsr-id-ref": nsr_id
},
4262 {"_admin.nsState": "NOT_INSTANTIATED"},
4264 except DbException
as e
:
4267 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4271 if operation_params
:
4272 autoremove
= operation_params
.get("autoremove", False)
4273 if nslcmop_operation_state
:
4275 await self
.msg
.aiowrite(
4280 "nslcmop_id": nslcmop_id
,
4281 "operationState": nslcmop_operation_state
,
4282 "autoremove": autoremove
,
4286 except Exception as e
:
4288 logging_text
+ "kafka_write notification Exception {}".format(e
)
4291 self
.logger
.debug(logging_text
+ "Exit")
4292 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4294 async def _wait_for_tasks(
4295 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4298 error_detail_list
= []
4300 pending_tasks
= list(created_tasks_info
.keys())
4301 num_tasks
= len(pending_tasks
)
4303 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4304 self
._write
_op
_status
(nslcmop_id
, stage
)
4305 while pending_tasks
:
4307 _timeout
= timeout
+ time_start
- time()
4308 done
, pending_tasks
= await asyncio
.wait(
4309 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4311 num_done
+= len(done
)
4312 if not done
: # Timeout
4313 for task
in pending_tasks
:
4314 new_error
= created_tasks_info
[task
] + ": Timeout"
4315 error_detail_list
.append(new_error
)
4316 error_list
.append(new_error
)
4319 if task
.cancelled():
4322 exc
= task
.exception()
4324 if isinstance(exc
, asyncio
.TimeoutError
):
4326 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4327 error_list
.append(created_tasks_info
[task
])
4328 error_detail_list
.append(new_error
)
4335 ROclient
.ROClientException
,
4341 self
.logger
.error(logging_text
+ new_error
)
4343 exc_traceback
= "".join(
4344 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4348 + created_tasks_info
[task
]
4354 logging_text
+ created_tasks_info
[task
] + ": Done"
4356 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4358 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4359 if nsr_id
: # update also nsr
4364 "errorDescription": "Error at: " + ", ".join(error_list
),
4365 "errorDetail": ". ".join(error_detail_list
),
4368 self
._write
_op
_status
(nslcmop_id
, stage
)
4369 return error_detail_list
4372 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4374 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4375 The default-value is used. If it is between < > it look for a value at instantiation_params
4376 :param primitive_desc: portion of VNFD/NSD that describes primitive
4377 :param params: Params provided by user
4378 :param instantiation_params: Instantiation params provided by user
4379 :return: a dictionary with the calculated params
4381 calculated_params
= {}
4382 for parameter
in primitive_desc
.get("parameter", ()):
4383 param_name
= parameter
["name"]
4384 if param_name
in params
:
4385 calculated_params
[param_name
] = params
[param_name
]
4386 elif "default-value" in parameter
or "value" in parameter
:
4387 if "value" in parameter
:
4388 calculated_params
[param_name
] = parameter
["value"]
4390 calculated_params
[param_name
] = parameter
["default-value"]
4392 isinstance(calculated_params
[param_name
], str)
4393 and calculated_params
[param_name
].startswith("<")
4394 and calculated_params
[param_name
].endswith(">")
4396 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4397 calculated_params
[param_name
] = instantiation_params
[
4398 calculated_params
[param_name
][1:-1]
4402 "Parameter {} needed to execute primitive {} not provided".format(
4403 calculated_params
[param_name
], primitive_desc
["name"]
4408 "Parameter {} needed to execute primitive {} not provided".format(
4409 param_name
, primitive_desc
["name"]
4413 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4414 calculated_params
[param_name
] = yaml
.safe_dump(
4415 calculated_params
[param_name
], default_flow_style
=True, width
=256
4417 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4419 ].startswith("!!yaml "):
4420 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4421 if parameter
.get("data-type") == "INTEGER":
4423 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4424 except ValueError: # error converting string to int
4426 "Parameter {} of primitive {} must be integer".format(
4427 param_name
, primitive_desc
["name"]
4430 elif parameter
.get("data-type") == "BOOLEAN":
4431 calculated_params
[param_name
] = not (
4432 (str(calculated_params
[param_name
])).lower() == "false"
4435 # add always ns_config_info if primitive name is config
4436 if primitive_desc
["name"] == "config":
4437 if "ns_config_info" in instantiation_params
:
4438 calculated_params
["ns_config_info"] = instantiation_params
[
4441 return calculated_params
4443 def _look_for_deployed_vca(
4450 ee_descriptor_id
=None,
4452 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4453 for vca
in deployed_vca
:
4456 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4459 vdu_count_index
is not None
4460 and vdu_count_index
!= vca
["vdu_count_index"]
4463 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4465 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4469 # vca_deployed not found
4471 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4472 " is not deployed".format(
4481 ee_id
= vca
.get("ee_id")
4483 "type", "lxc_proxy_charm"
4484 ) # default value for backward compatibility - proxy charm
4487 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4488 "execution environment".format(
4489 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4492 return ee_id
, vca_type
4494 async def _ns_execute_primitive(
4500 retries_interval
=30,
4507 if primitive
== "config":
4508 primitive_params
= {"params": primitive_params
}
4510 vca_type
= vca_type
or "lxc_proxy_charm"
4514 output
= await asyncio
.wait_for(
4515 self
.vca_map
[vca_type
].exec_primitive(
4517 primitive_name
=primitive
,
4518 params_dict
=primitive_params
,
4519 progress_timeout
=self
.timeout_progress_primitive
,
4520 total_timeout
=self
.timeout_primitive
,
4524 timeout
=timeout
or self
.timeout_primitive
,
4528 except asyncio
.CancelledError
:
4530 except Exception as e
: # asyncio.TimeoutError
4531 if isinstance(e
, asyncio
.TimeoutError
):
4536 "Error executing action {} on {} -> {}".format(
4541 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4543 return "FAILED", str(e
)
4545 return "COMPLETED", output
4547 except (LcmException
, asyncio
.CancelledError
):
4549 except Exception as e
:
4550 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4552 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4554 Updating the vca_status with latest juju information in nsrs record
4555 :param: nsr_id: Id of the nsr
4556 :param: nslcmop_id: Id of the nslcmop
4560 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4561 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4562 vca_id
= self
.get_vca_id({}, db_nsr
)
4563 if db_nsr
["_admin"]["deployed"]["K8s"]:
4564 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4565 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4566 await self
._on
_update
_k
8s
_db
(
4567 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4570 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4571 table
, filter = "nsrs", {"_id": nsr_id
}
4572 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4573 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4575 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4576 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4578 async def action(self
, nsr_id
, nslcmop_id
):
4579 # Try to lock HA task here
4580 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4581 if not task_is_locked_by_me
:
4584 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4585 self
.logger
.debug(logging_text
+ "Enter")
4586 # get all needed from database
4590 db_nslcmop_update
= {}
4591 nslcmop_operation_state
= None
4592 error_description_nslcmop
= None
4595 # wait for any previous tasks in process
4596 step
= "Waiting for previous operations to terminate"
4597 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4599 self
._write
_ns
_status
(
4602 current_operation
="RUNNING ACTION",
4603 current_operation_id
=nslcmop_id
,
4606 step
= "Getting information from database"
4607 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4608 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4610 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4611 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4612 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4613 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4614 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4615 primitive
= db_nslcmop
["operationParams"]["primitive"]
4616 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4617 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4618 "timeout_ns_action", self
.timeout_primitive
4622 step
= "Getting vnfr from database"
4623 db_vnfr
= self
.db
.get_one(
4624 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4626 step
= "Getting vnfd from database"
4627 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4629 step
= "Getting nsd from database"
4630 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4632 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4633 # for backward compatibility
4634 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4635 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4636 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4637 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4639 # look for primitive
4640 config_primitive_desc
= descriptor_configuration
= None
4642 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4644 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4646 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4648 descriptor_configuration
= db_nsd
.get("ns-configuration")
4650 if descriptor_configuration
and descriptor_configuration
.get(
4653 for config_primitive
in descriptor_configuration
["config-primitive"]:
4654 if config_primitive
["name"] == primitive
:
4655 config_primitive_desc
= config_primitive
4658 if not config_primitive_desc
:
4659 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4661 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4665 primitive_name
= primitive
4666 ee_descriptor_id
= None
4668 primitive_name
= config_primitive_desc
.get(
4669 "execution-environment-primitive", primitive
4671 ee_descriptor_id
= config_primitive_desc
.get(
4672 "execution-environment-ref"
4678 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4680 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4683 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4685 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4687 desc_params
= parse_yaml_strings(
4688 db_vnfr
.get("additionalParamsForVnf")
4691 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4692 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4693 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4695 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4696 actions
.add(primitive
["name"])
4697 for primitive
in kdu_configuration
.get("config-primitive", []):
4698 actions
.add(primitive
["name"])
4699 kdu_action
= True if primitive_name
in actions
else False
4701 # TODO check if ns is in a proper status
4703 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4705 # kdur and desc_params already set from before
4706 if primitive_params
:
4707 desc_params
.update(primitive_params
)
4708 # TODO Check if we will need something at vnf level
4709 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4711 kdu_name
== kdu
["kdu-name"]
4712 and kdu
["member-vnf-index"] == vnf_index
4717 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4720 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4721 msg
= "unknown k8scluster-type '{}'".format(
4722 kdu
.get("k8scluster-type")
4724 raise LcmException(msg
)
4727 "collection": "nsrs",
4728 "filter": {"_id": nsr_id
},
4729 "path": "_admin.deployed.K8s.{}".format(index
),
4733 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4735 step
= "Executing kdu {}".format(primitive_name
)
4736 if primitive_name
== "upgrade":
4737 if desc_params
.get("kdu_model"):
4738 kdu_model
= desc_params
.get("kdu_model")
4739 del desc_params
["kdu_model"]
4741 kdu_model
= kdu
.get("kdu-model")
4742 parts
= kdu_model
.split(sep
=":")
4744 kdu_model
= parts
[0]
4746 detailed_status
= await asyncio
.wait_for(
4747 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4748 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4749 kdu_instance
=kdu
.get("kdu-instance"),
4751 kdu_model
=kdu_model
,
4754 timeout
=timeout_ns_action
,
4756 timeout
=timeout_ns_action
+ 10,
4759 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4761 elif primitive_name
== "rollback":
4762 detailed_status
= await asyncio
.wait_for(
4763 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4764 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4765 kdu_instance
=kdu
.get("kdu-instance"),
4768 timeout
=timeout_ns_action
,
4770 elif primitive_name
== "status":
4771 detailed_status
= await asyncio
.wait_for(
4772 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4773 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4774 kdu_instance
=kdu
.get("kdu-instance"),
4777 timeout
=timeout_ns_action
,
4780 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
4781 kdu
["kdu-name"], nsr_id
4783 params
= self
._map
_primitive
_params
(
4784 config_primitive_desc
, primitive_params
, desc_params
4787 detailed_status
= await asyncio
.wait_for(
4788 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
4789 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4790 kdu_instance
=kdu_instance
,
4791 primitive_name
=primitive_name
,
4794 timeout
=timeout_ns_action
,
4797 timeout
=timeout_ns_action
,
4801 nslcmop_operation_state
= "COMPLETED"
4803 detailed_status
= ""
4804 nslcmop_operation_state
= "FAILED"
4806 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
4807 nsr_deployed
["VCA"],
4808 member_vnf_index
=vnf_index
,
4810 vdu_count_index
=vdu_count_index
,
4811 ee_descriptor_id
=ee_descriptor_id
,
4813 for vca_index
, vca_deployed
in enumerate(
4814 db_nsr
["_admin"]["deployed"]["VCA"]
4816 if vca_deployed
.get("member-vnf-index") == vnf_index
:
4818 "collection": "nsrs",
4819 "filter": {"_id": nsr_id
},
4820 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
4824 nslcmop_operation_state
,
4826 ) = await self
._ns
_execute
_primitive
(
4828 primitive
=primitive_name
,
4829 primitive_params
=self
._map
_primitive
_params
(
4830 config_primitive_desc
, primitive_params
, desc_params
4832 timeout
=timeout_ns_action
,
4838 db_nslcmop_update
["detailed-status"] = detailed_status
4839 error_description_nslcmop
= (
4840 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
4844 + " task Done with result {} {}".format(
4845 nslcmop_operation_state
, detailed_status
4848 return # database update is called inside finally
4850 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
4851 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4853 except asyncio
.CancelledError
:
4855 logging_text
+ "Cancelled Exception while '{}'".format(step
)
4857 exc
= "Operation was cancelled"
4858 except asyncio
.TimeoutError
:
4859 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
4861 except Exception as e
:
4862 exc
= traceback
.format_exc()
4863 self
.logger
.critical(
4864 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
4873 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
4874 nslcmop_operation_state
= "FAILED"
4876 self
._write
_ns
_status
(
4880 ], # TODO check if degraded. For the moment use previous status
4881 current_operation
="IDLE",
4882 current_operation_id
=None,
4883 # error_description=error_description_nsr,
4884 # error_detail=error_detail,
4885 other_update
=db_nsr_update
,
4888 self
._write
_op
_status
(
4891 error_message
=error_description_nslcmop
,
4892 operation_state
=nslcmop_operation_state
,
4893 other_update
=db_nslcmop_update
,
4896 if nslcmop_operation_state
:
4898 await self
.msg
.aiowrite(
4903 "nslcmop_id": nslcmop_id
,
4904 "operationState": nslcmop_operation_state
,
4908 except Exception as e
:
4910 logging_text
+ "kafka_write notification Exception {}".format(e
)
4912 self
.logger
.debug(logging_text
+ "Exit")
4913 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
4914 return nslcmop_operation_state
, detailed_status
4916 async def scale(self
, nsr_id
, nslcmop_id
):
4917 # Try to lock HA task here
4918 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4919 if not task_is_locked_by_me
:
4922 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
4923 stage
= ["", "", ""]
4924 tasks_dict_info
= {}
4925 # ^ stage, step, VIM progress
4926 self
.logger
.debug(logging_text
+ "Enter")
4927 # get all needed from database
4929 db_nslcmop_update
= {}
4932 # in case of error, indicates what part of scale was failed to put nsr at error status
4933 scale_process
= None
4934 old_operational_status
= ""
4935 old_config_status
= ""
4938 # wait for any previous tasks in process
4939 step
= "Waiting for previous operations to terminate"
4940 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4941 self
._write
_ns
_status
(
4944 current_operation
="SCALING",
4945 current_operation_id
=nslcmop_id
,
4948 step
= "Getting nslcmop from database"
4950 step
+ " after having waited for previous tasks to be completed"
4952 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4954 step
= "Getting nsr from database"
4955 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4956 old_operational_status
= db_nsr
["operational-status"]
4957 old_config_status
= db_nsr
["config-status"]
4959 step
= "Parsing scaling parameters"
4960 db_nsr_update
["operational-status"] = "scaling"
4961 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4962 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4964 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
4966 ]["member-vnf-index"]
4967 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
4969 ]["scaling-group-descriptor"]
4970 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
4971 # for backward compatibility
4972 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4973 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4974 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4975 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4977 step
= "Getting vnfr from database"
4978 db_vnfr
= self
.db
.get_one(
4979 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4982 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4984 step
= "Getting vnfd from database"
4985 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4987 base_folder
= db_vnfd
["_admin"]["storage"]
4989 step
= "Getting scaling-group-descriptor"
4990 scaling_descriptor
= find_in_list(
4991 get_scaling_aspect(db_vnfd
),
4992 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
4994 if not scaling_descriptor
:
4996 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
4997 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5000 step
= "Sending scale order to VIM"
5001 # TODO check if ns is in a proper status
5003 if not db_nsr
["_admin"].get("scaling-group"):
5008 "_admin.scaling-group": [
5009 {"name": scaling_group
, "nb-scale-op": 0}
5013 admin_scale_index
= 0
5015 for admin_scale_index
, admin_scale_info
in enumerate(
5016 db_nsr
["_admin"]["scaling-group"]
5018 if admin_scale_info
["name"] == scaling_group
:
5019 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5021 else: # not found, set index one plus last element and add new entry with the name
5022 admin_scale_index
+= 1
5024 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5027 vca_scaling_info
= []
5028 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5029 if scaling_type
== "SCALE_OUT":
5030 if "aspect-delta-details" not in scaling_descriptor
:
5032 "Aspect delta details not fount in scaling descriptor {}".format(
5033 scaling_descriptor
["name"]
5036 # count if max-instance-count is reached
5037 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5039 scaling_info
["scaling_direction"] = "OUT"
5040 scaling_info
["vdu-create"] = {}
5041 scaling_info
["kdu-create"] = {}
5042 for delta
in deltas
:
5043 for vdu_delta
in delta
.get("vdu-delta", {}):
5044 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5045 # vdu_index also provides the number of instance of the targeted vdu
5046 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5047 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5051 additional_params
= (
5052 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5055 cloud_init_list
= []
5057 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5058 max_instance_count
= 10
5059 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5060 max_instance_count
= vdu_profile
.get(
5061 "max-number-of-instances", 10
5064 default_instance_num
= get_number_of_instances(
5067 instances_number
= vdu_delta
.get("number-of-instances", 1)
5068 nb_scale_op
+= instances_number
5070 new_instance_count
= nb_scale_op
+ default_instance_num
5071 # Control if new count is over max and vdu count is less than max.
5072 # Then assign new instance count
5073 if new_instance_count
> max_instance_count
> vdu_count
:
5074 instances_number
= new_instance_count
- max_instance_count
5076 instances_number
= instances_number
5078 if new_instance_count
> max_instance_count
:
5080 "reached the limit of {} (max-instance-count) "
5081 "scaling-out operations for the "
5082 "scaling-group-descriptor '{}'".format(
5083 nb_scale_op
, scaling_group
5086 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5088 # TODO Information of its own ip is not available because db_vnfr is not updated.
5089 additional_params
["OSM"] = get_osm_params(
5090 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5092 cloud_init_list
.append(
5093 self
._parse
_cloud
_init
(
5100 vca_scaling_info
.append(
5102 "osm_vdu_id": vdu_delta
["id"],
5103 "member-vnf-index": vnf_index
,
5105 "vdu_index": vdu_index
+ x
,
5108 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5109 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5110 kdu_profile
= get_kdu_profile(db_vnfd
, kdu_delta
["id"])
5111 kdu_name
= kdu_profile
["kdu-name"]
5112 resource_name
= kdu_profile
["resource-name"]
5114 # Might have different kdus in the same delta
5115 # Should have list for each kdu
5116 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5117 scaling_info
["kdu-create"][kdu_name
] = []
5119 kdur
= get_kdur(db_vnfr
, kdu_name
)
5120 if kdur
.get("helm-chart"):
5121 k8s_cluster_type
= "helm-chart-v3"
5122 self
.logger
.debug("kdur: {}".format(kdur
))
5124 kdur
.get("helm-version")
5125 and kdur
.get("helm-version") == "v2"
5127 k8s_cluster_type
= "helm-chart"
5128 raise NotImplementedError
5129 elif kdur
.get("juju-bundle"):
5130 k8s_cluster_type
= "juju-bundle"
5133 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5134 "juju-bundle. Maybe an old NBI version is running".format(
5135 db_vnfr
["member-vnf-index-ref"], kdu_name
5139 max_instance_count
= 10
5140 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5141 max_instance_count
= kdu_profile
.get(
5142 "max-number-of-instances", 10
5145 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5146 deployed_kdu
, _
= get_deployed_kdu(
5147 nsr_deployed
, kdu_name
, vnf_index
5149 if deployed_kdu
is None:
5151 "KDU '{}' for vnf '{}' not deployed".format(
5155 kdu_instance
= deployed_kdu
.get("kdu-instance")
5156 instance_num
= await self
.k8scluster_map
[
5158 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5159 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5160 "number-of-instances", 1
5163 # Control if new count is over max and instance_num is less than max.
5164 # Then assign max instance number to kdu replica count
5165 if kdu_replica_count
> max_instance_count
> instance_num
:
5166 kdu_replica_count
= max_instance_count
5167 if kdu_replica_count
> max_instance_count
:
5169 "reached the limit of {} (max-instance-count) "
5170 "scaling-out operations for the "
5171 "scaling-group-descriptor '{}'".format(
5172 instance_num
, scaling_group
5176 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5177 vca_scaling_info
.append(
5179 "osm_kdu_id": kdu_name
,
5180 "member-vnf-index": vnf_index
,
5182 "kdu_index": instance_num
+ x
- 1,
5185 scaling_info
["kdu-create"][kdu_name
].append(
5187 "member-vnf-index": vnf_index
,
5189 "k8s-cluster-type": k8s_cluster_type
,
5190 "resource-name": resource_name
,
5191 "scale": kdu_replica_count
,
5194 elif scaling_type
== "SCALE_IN":
5195 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5197 scaling_info
["scaling_direction"] = "IN"
5198 scaling_info
["vdu-delete"] = {}
5199 scaling_info
["kdu-delete"] = {}
5201 for delta
in deltas
:
5202 for vdu_delta
in delta
.get("vdu-delta", {}):
5203 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5204 min_instance_count
= 0
5205 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5206 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5207 min_instance_count
= vdu_profile
["min-number-of-instances"]
5209 default_instance_num
= get_number_of_instances(
5210 db_vnfd
, vdu_delta
["id"]
5212 instance_num
= vdu_delta
.get("number-of-instances", 1)
5213 nb_scale_op
-= instance_num
5215 new_instance_count
= nb_scale_op
+ default_instance_num
5217 if new_instance_count
< min_instance_count
< vdu_count
:
5218 instances_number
= min_instance_count
- new_instance_count
5220 instances_number
= instance_num
5222 if new_instance_count
< min_instance_count
:
5224 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5225 "scaling-group-descriptor '{}'".format(
5226 nb_scale_op
, scaling_group
5229 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5230 vca_scaling_info
.append(
5232 "osm_vdu_id": vdu_delta
["id"],
5233 "member-vnf-index": vnf_index
,
5235 "vdu_index": vdu_index
- 1 - x
,
5238 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5239 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5240 kdu_profile
= get_kdu_profile(db_vnfd
, kdu_delta
["id"])
5241 kdu_name
= kdu_profile
["kdu-name"]
5242 resource_name
= kdu_profile
["resource-name"]
5244 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5245 scaling_info
["kdu-delete"][kdu_name
] = []
5247 kdur
= get_kdur(db_vnfr
, kdu_name
)
5248 if kdur
.get("helm-chart"):
5249 k8s_cluster_type
= "helm-chart-v3"
5250 self
.logger
.debug("kdur: {}".format(kdur
))
5252 kdur
.get("helm-version")
5253 and kdur
.get("helm-version") == "v2"
5255 k8s_cluster_type
= "helm-chart"
5256 raise NotImplementedError
5257 elif kdur
.get("juju-bundle"):
5258 k8s_cluster_type
= "juju-bundle"
5261 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5262 "juju-bundle. Maybe an old NBI version is running".format(
5263 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5267 min_instance_count
= 0
5268 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5269 min_instance_count
= kdu_profile
["min-number-of-instances"]
5271 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5272 deployed_kdu
, _
= get_deployed_kdu(
5273 nsr_deployed
, kdu_name
, vnf_index
5275 if deployed_kdu
is None:
5277 "KDU '{}' for vnf '{}' not deployed".format(
5281 kdu_instance
= deployed_kdu
.get("kdu-instance")
5282 instance_num
= await self
.k8scluster_map
[
5284 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5285 kdu_replica_count
= instance_num
- kdu_delta
.get(
5286 "number-of-instances", 1
5289 if kdu_replica_count
< min_instance_count
< instance_num
:
5290 kdu_replica_count
= min_instance_count
5291 if kdu_replica_count
< min_instance_count
:
5293 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5294 "scaling-group-descriptor '{}'".format(
5295 instance_num
, scaling_group
5299 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5300 vca_scaling_info
.append(
5302 "osm_kdu_id": kdu_name
,
5303 "member-vnf-index": vnf_index
,
5305 "kdu_index": instance_num
- x
- 1,
5308 scaling_info
["kdu-delete"][kdu_name
].append(
5310 "member-vnf-index": vnf_index
,
5312 "k8s-cluster-type": k8s_cluster_type
,
5313 "resource-name": resource_name
,
5314 "scale": kdu_replica_count
,
5318 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5319 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5320 if scaling_info
["scaling_direction"] == "IN":
5321 for vdur
in reversed(db_vnfr
["vdur"]):
5322 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5323 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5324 scaling_info
["vdu"].append(
5326 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5327 "vdu_id": vdur
["vdu-id-ref"],
5331 for interface
in vdur
["interfaces"]:
5332 scaling_info
["vdu"][-1]["interface"].append(
5334 "name": interface
["name"],
5335 "ip_address": interface
["ip-address"],
5336 "mac_address": interface
.get("mac-address"),
5339 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5342 step
= "Executing pre-scale vnf-config-primitive"
5343 if scaling_descriptor
.get("scaling-config-action"):
5344 for scaling_config_action
in scaling_descriptor
[
5345 "scaling-config-action"
5348 scaling_config_action
.get("trigger") == "pre-scale-in"
5349 and scaling_type
== "SCALE_IN"
5351 scaling_config_action
.get("trigger") == "pre-scale-out"
5352 and scaling_type
== "SCALE_OUT"
5354 vnf_config_primitive
= scaling_config_action
[
5355 "vnf-config-primitive-name-ref"
5357 step
= db_nslcmop_update
[
5359 ] = "executing pre-scale scaling-config-action '{}'".format(
5360 vnf_config_primitive
5363 # look for primitive
5364 for config_primitive
in (
5365 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5366 ).get("config-primitive", ()):
5367 if config_primitive
["name"] == vnf_config_primitive
:
5371 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5372 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5373 "primitive".format(scaling_group
, vnf_config_primitive
)
5376 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5377 if db_vnfr
.get("additionalParamsForVnf"):
5378 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5380 scale_process
= "VCA"
5381 db_nsr_update
["config-status"] = "configuring pre-scaling"
5382 primitive_params
= self
._map
_primitive
_params
(
5383 config_primitive
, {}, vnfr_params
5386 # Pre-scale retry check: Check if this sub-operation has been executed before
5387 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5390 vnf_config_primitive
,
5394 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5395 # Skip sub-operation
5396 result
= "COMPLETED"
5397 result_detail
= "Done"
5400 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5401 vnf_config_primitive
, result
, result_detail
5405 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5406 # New sub-operation: Get index of this sub-operation
5408 len(db_nslcmop
.get("_admin", {}).get("operations"))
5413 + "vnf_config_primitive={} New sub-operation".format(
5414 vnf_config_primitive
5418 # retry: Get registered params for this existing sub-operation
5419 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5422 vnf_index
= op
.get("member_vnf_index")
5423 vnf_config_primitive
= op
.get("primitive")
5424 primitive_params
= op
.get("primitive_params")
5427 + "vnf_config_primitive={} Sub-operation retry".format(
5428 vnf_config_primitive
5431 # Execute the primitive, either with new (first-time) or registered (reintent) args
5432 ee_descriptor_id
= config_primitive
.get(
5433 "execution-environment-ref"
5435 primitive_name
= config_primitive
.get(
5436 "execution-environment-primitive", vnf_config_primitive
5438 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5439 nsr_deployed
["VCA"],
5440 member_vnf_index
=vnf_index
,
5442 vdu_count_index
=None,
5443 ee_descriptor_id
=ee_descriptor_id
,
5445 result
, result_detail
= await self
._ns
_execute
_primitive
(
5454 + "vnf_config_primitive={} Done with result {} {}".format(
5455 vnf_config_primitive
, result
, result_detail
5458 # Update operationState = COMPLETED | FAILED
5459 self
._update
_suboperation
_status
(
5460 db_nslcmop
, op_index
, result
, result_detail
5463 if result
== "FAILED":
5464 raise LcmException(result_detail
)
5465 db_nsr_update
["config-status"] = old_config_status
5466 scale_process
= None
5470 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5473 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5476 # SCALE-IN VCA - BEGIN
5477 if vca_scaling_info
:
5478 step
= db_nslcmop_update
[
5480 ] = "Deleting the execution environments"
5481 scale_process
= "VCA"
5482 for vca_info
in vca_scaling_info
:
5483 if vca_info
["type"] == "delete":
5484 member_vnf_index
= str(vca_info
["member-vnf-index"])
5486 logging_text
+ "vdu info: {}".format(vca_info
)
5488 if vca_info
.get("osm_vdu_id"):
5489 vdu_id
= vca_info
["osm_vdu_id"]
5490 vdu_index
= int(vca_info
["vdu_index"])
5493 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5494 member_vnf_index
, vdu_id
, vdu_index
5498 kdu_id
= vca_info
["osm_kdu_id"]
5501 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5502 member_vnf_index
, kdu_id
, vdu_index
5504 stage
[2] = step
= "Scaling in VCA"
5505 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5506 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5507 config_update
= db_nsr
["configurationStatus"]
5508 for vca_index
, vca
in enumerate(vca_update
):
5510 (vca
or vca
.get("ee_id"))
5511 and vca
["member-vnf-index"] == member_vnf_index
5512 and vca
["vdu_count_index"] == vdu_index
5514 if vca
.get("vdu_id"):
5515 config_descriptor
= get_configuration(
5516 db_vnfd
, vca
.get("vdu_id")
5518 elif vca
.get("kdu_name"):
5519 config_descriptor
= get_configuration(
5520 db_vnfd
, vca
.get("kdu_name")
5523 config_descriptor
= get_configuration(
5524 db_vnfd
, db_vnfd
["id"]
5526 operation_params
= (
5527 db_nslcmop
.get("operationParams") or {}
5529 exec_terminate_primitives
= not operation_params
.get(
5530 "skip_terminate_primitives"
5531 ) and vca
.get("needed_terminate")
5532 task
= asyncio
.ensure_future(
5541 exec_primitives
=exec_terminate_primitives
,
5545 timeout
=self
.timeout_charm_delete
,
5548 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5551 del vca_update
[vca_index
]
5552 del config_update
[vca_index
]
5553 # wait for pending tasks of terminate primitives
5557 + "Waiting for tasks {}".format(
5558 list(tasks_dict_info
.keys())
5561 error_list
= await self
._wait
_for
_tasks
(
5565 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5570 tasks_dict_info
.clear()
5572 raise LcmException("; ".join(error_list
))
5574 db_vca_and_config_update
= {
5575 "_admin.deployed.VCA": vca_update
,
5576 "configurationStatus": config_update
,
5579 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5581 scale_process
= None
5582 # SCALE-IN VCA - END
5585 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5586 scale_process
= "RO"
5587 if self
.ro_config
.get("ng"):
5588 await self
._scale
_ng
_ro
(
5589 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5591 scaling_info
.pop("vdu-create", None)
5592 scaling_info
.pop("vdu-delete", None)
5594 scale_process
= None
5598 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5599 scale_process
= "KDU"
5600 await self
._scale
_kdu
(
5601 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5603 scaling_info
.pop("kdu-create", None)
5604 scaling_info
.pop("kdu-delete", None)
5606 scale_process
= None
5610 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5612 # SCALE-UP VCA - BEGIN
5613 if vca_scaling_info
:
5614 step
= db_nslcmop_update
[
5616 ] = "Creating new execution environments"
5617 scale_process
= "VCA"
5618 for vca_info
in vca_scaling_info
:
5619 if vca_info
["type"] == "create":
5620 member_vnf_index
= str(vca_info
["member-vnf-index"])
5622 logging_text
+ "vdu info: {}".format(vca_info
)
5624 vnfd_id
= db_vnfr
["vnfd-ref"]
5625 if vca_info
.get("osm_vdu_id"):
5626 vdu_index
= int(vca_info
["vdu_index"])
5627 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5628 if db_vnfr
.get("additionalParamsForVnf"):
5629 deploy_params
.update(
5631 db_vnfr
["additionalParamsForVnf"].copy()
5634 descriptor_config
= get_configuration(
5635 db_vnfd
, db_vnfd
["id"]
5637 if descriptor_config
:
5642 logging_text
=logging_text
5643 + "member_vnf_index={} ".format(member_vnf_index
),
5646 nslcmop_id
=nslcmop_id
,
5652 member_vnf_index
=member_vnf_index
,
5653 vdu_index
=vdu_index
,
5655 deploy_params
=deploy_params
,
5656 descriptor_config
=descriptor_config
,
5657 base_folder
=base_folder
,
5658 task_instantiation_info
=tasks_dict_info
,
5661 vdu_id
= vca_info
["osm_vdu_id"]
5662 vdur
= find_in_list(
5663 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5665 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5666 if vdur
.get("additionalParams"):
5667 deploy_params_vdu
= parse_yaml_strings(
5668 vdur
["additionalParams"]
5671 deploy_params_vdu
= deploy_params
5672 deploy_params_vdu
["OSM"] = get_osm_params(
5673 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5675 if descriptor_config
:
5680 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5681 member_vnf_index
, vdu_id
, vdu_index
5683 stage
[2] = step
= "Scaling out VCA"
5684 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5686 logging_text
=logging_text
5687 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5688 member_vnf_index
, vdu_id
, vdu_index
5692 nslcmop_id
=nslcmop_id
,
5698 member_vnf_index
=member_vnf_index
,
5699 vdu_index
=vdu_index
,
5701 deploy_params
=deploy_params_vdu
,
5702 descriptor_config
=descriptor_config
,
5703 base_folder
=base_folder
,
5704 task_instantiation_info
=tasks_dict_info
,
5708 kdu_name
= vca_info
["osm_kdu_id"]
5709 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5710 if descriptor_config
:
5712 kdu_index
= int(vca_info
["kdu_index"])
5716 for x
in db_vnfr
["kdur"]
5717 if x
["kdu-name"] == kdu_name
5719 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5720 if kdur
.get("additionalParams"):
5721 deploy_params_kdu
= parse_yaml_strings(
5722 kdur
["additionalParams"]
5726 logging_text
=logging_text
,
5729 nslcmop_id
=nslcmop_id
,
5735 member_vnf_index
=member_vnf_index
,
5736 vdu_index
=kdu_index
,
5738 deploy_params
=deploy_params_kdu
,
5739 descriptor_config
=descriptor_config
,
5740 base_folder
=base_folder
,
5741 task_instantiation_info
=tasks_dict_info
,
5744 # SCALE-UP VCA - END
5745 scale_process
= None
5748 # execute primitive service POST-SCALING
5749 step
= "Executing post-scale vnf-config-primitive"
5750 if scaling_descriptor
.get("scaling-config-action"):
5751 for scaling_config_action
in scaling_descriptor
[
5752 "scaling-config-action"
5755 scaling_config_action
.get("trigger") == "post-scale-in"
5756 and scaling_type
== "SCALE_IN"
5758 scaling_config_action
.get("trigger") == "post-scale-out"
5759 and scaling_type
== "SCALE_OUT"
5761 vnf_config_primitive
= scaling_config_action
[
5762 "vnf-config-primitive-name-ref"
5764 step
= db_nslcmop_update
[
5766 ] = "executing post-scale scaling-config-action '{}'".format(
5767 vnf_config_primitive
5770 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5771 if db_vnfr
.get("additionalParamsForVnf"):
5772 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5774 # look for primitive
5775 for config_primitive
in (
5776 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5777 ).get("config-primitive", ()):
5778 if config_primitive
["name"] == vnf_config_primitive
:
5782 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
5783 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
5784 "config-primitive".format(
5785 scaling_group
, vnf_config_primitive
5788 scale_process
= "VCA"
5789 db_nsr_update
["config-status"] = "configuring post-scaling"
5790 primitive_params
= self
._map
_primitive
_params
(
5791 config_primitive
, {}, vnfr_params
5794 # Post-scale retry check: Check if this sub-operation has been executed before
5795 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5798 vnf_config_primitive
,
5802 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5803 # Skip sub-operation
5804 result
= "COMPLETED"
5805 result_detail
= "Done"
5808 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5809 vnf_config_primitive
, result
, result_detail
5813 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5814 # New sub-operation: Get index of this sub-operation
5816 len(db_nslcmop
.get("_admin", {}).get("operations"))
5821 + "vnf_config_primitive={} New sub-operation".format(
5822 vnf_config_primitive
5826 # retry: Get registered params for this existing sub-operation
5827 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5830 vnf_index
= op
.get("member_vnf_index")
5831 vnf_config_primitive
= op
.get("primitive")
5832 primitive_params
= op
.get("primitive_params")
5835 + "vnf_config_primitive={} Sub-operation retry".format(
5836 vnf_config_primitive
5839 # Execute the primitive, either with new (first-time) or registered (reintent) args
5840 ee_descriptor_id
= config_primitive
.get(
5841 "execution-environment-ref"
5843 primitive_name
= config_primitive
.get(
5844 "execution-environment-primitive", vnf_config_primitive
5846 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5847 nsr_deployed
["VCA"],
5848 member_vnf_index
=vnf_index
,
5850 vdu_count_index
=None,
5851 ee_descriptor_id
=ee_descriptor_id
,
5853 result
, result_detail
= await self
._ns
_execute
_primitive
(
5862 + "vnf_config_primitive={} Done with result {} {}".format(
5863 vnf_config_primitive
, result
, result_detail
5866 # Update operationState = COMPLETED | FAILED
5867 self
._update
_suboperation
_status
(
5868 db_nslcmop
, op_index
, result
, result_detail
5871 if result
== "FAILED":
5872 raise LcmException(result_detail
)
5873 db_nsr_update
["config-status"] = old_config_status
5874 scale_process
= None
5879 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
5880 db_nsr_update
["operational-status"] = (
5882 if old_operational_status
== "failed"
5883 else old_operational_status
5885 db_nsr_update
["config-status"] = old_config_status
5888 ROclient
.ROClientException
,
5893 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5895 except asyncio
.CancelledError
:
5897 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5899 exc
= "Operation was cancelled"
5900 except Exception as e
:
5901 exc
= traceback
.format_exc()
5902 self
.logger
.critical(
5903 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5907 self
._write
_ns
_status
(
5910 current_operation
="IDLE",
5911 current_operation_id
=None,
5914 stage
[1] = "Waiting for instantiate pending tasks."
5915 self
.logger
.debug(logging_text
+ stage
[1])
5916 exc
= await self
._wait
_for
_tasks
(
5919 self
.timeout_ns_deploy
,
5927 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5928 nslcmop_operation_state
= "FAILED"
5930 db_nsr_update
["operational-status"] = old_operational_status
5931 db_nsr_update
["config-status"] = old_config_status
5932 db_nsr_update
["detailed-status"] = ""
5934 if "VCA" in scale_process
:
5935 db_nsr_update
["config-status"] = "failed"
5936 if "RO" in scale_process
:
5937 db_nsr_update
["operational-status"] = "failed"
5940 ] = "FAILED scaling nslcmop={} {}: {}".format(
5941 nslcmop_id
, step
, exc
5944 error_description_nslcmop
= None
5945 nslcmop_operation_state
= "COMPLETED"
5946 db_nslcmop_update
["detailed-status"] = "Done"
5948 self
._write
_op
_status
(
5951 error_message
=error_description_nslcmop
,
5952 operation_state
=nslcmop_operation_state
,
5953 other_update
=db_nslcmop_update
,
5956 self
._write
_ns
_status
(
5959 current_operation
="IDLE",
5960 current_operation_id
=None,
5961 other_update
=db_nsr_update
,
5964 if nslcmop_operation_state
:
5968 "nslcmop_id": nslcmop_id
,
5969 "operationState": nslcmop_operation_state
,
5971 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
5972 except Exception as e
:
5974 logging_text
+ "kafka_write notification Exception {}".format(e
)
5976 self
.logger
.debug(logging_text
+ "Exit")
5977 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
5979 async def _scale_kdu(
5980 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5982 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
5983 for kdu_name
in _scaling_info
:
5984 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
5985 deployed_kdu
, index
= get_deployed_kdu(
5986 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
5988 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
5989 kdu_instance
= deployed_kdu
["kdu-instance"]
5990 scale
= int(kdu_scaling_info
["scale"])
5991 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
5994 "collection": "nsrs",
5995 "filter": {"_id": nsr_id
},
5996 "path": "_admin.deployed.K8s.{}".format(index
),
5999 step
= "scaling application {}".format(
6000 kdu_scaling_info
["resource-name"]
6002 self
.logger
.debug(logging_text
+ step
)
6004 if kdu_scaling_info
["type"] == "delete":
6005 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6008 and kdu_config
.get("terminate-config-primitive")
6009 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6011 terminate_config_primitive_list
= kdu_config
.get(
6012 "terminate-config-primitive"
6014 terminate_config_primitive_list
.sort(
6015 key
=lambda val
: int(val
["seq"])
6019 terminate_config_primitive
6020 ) in terminate_config_primitive_list
:
6021 primitive_params_
= self
._map
_primitive
_params
(
6022 terminate_config_primitive
, {}, {}
6024 step
= "execute terminate config primitive"
6025 self
.logger
.debug(logging_text
+ step
)
6026 await asyncio
.wait_for(
6027 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6028 cluster_uuid
=cluster_uuid
,
6029 kdu_instance
=kdu_instance
,
6030 primitive_name
=terminate_config_primitive
["name"],
6031 params
=primitive_params_
,
6038 await asyncio
.wait_for(
6039 self
.k8scluster_map
[k8s_cluster_type
].scale(
6042 kdu_scaling_info
["resource-name"],
6045 timeout
=self
.timeout_vca_on_error
,
6048 if kdu_scaling_info
["type"] == "create":
6049 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6052 and kdu_config
.get("initial-config-primitive")
6053 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6055 initial_config_primitive_list
= kdu_config
.get(
6056 "initial-config-primitive"
6058 initial_config_primitive_list
.sort(
6059 key
=lambda val
: int(val
["seq"])
6062 for initial_config_primitive
in initial_config_primitive_list
:
6063 primitive_params_
= self
._map
_primitive
_params
(
6064 initial_config_primitive
, {}, {}
6066 step
= "execute initial config primitive"
6067 self
.logger
.debug(logging_text
+ step
)
6068 await asyncio
.wait_for(
6069 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6070 cluster_uuid
=cluster_uuid
,
6071 kdu_instance
=kdu_instance
,
6072 primitive_name
=initial_config_primitive
["name"],
6073 params
=primitive_params_
,
6080 async def _scale_ng_ro(
6081 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6083 nsr_id
= db_nslcmop
["nsInstanceId"]
6084 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6087 # read from db: vnfd's for every vnf
6090 # for each vnf in ns, read vnfd
6091 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6092 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6093 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6094 # if we haven't this vnfd, read it from db
6095 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6097 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6098 db_vnfds
.append(vnfd
)
6099 n2vc_key
= self
.n2vc
.get_public_key()
6100 n2vc_key_list
= [n2vc_key
]
6103 vdu_scaling_info
.get("vdu-create"),
6104 vdu_scaling_info
.get("vdu-delete"),
6107 # db_vnfr has been updated, update db_vnfrs to use it
6108 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6109 await self
._instantiate
_ng
_ro
(
6119 start_deploy
=time(),
6120 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6122 if vdu_scaling_info
.get("vdu-delete"):
6124 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6127 async def add_prometheus_metrics(
6128 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6130 if not self
.prometheus
:
6132 # look if exist a file called 'prometheus*.j2' and
6133 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6137 for f
in artifact_content
6138 if f
.startswith("prometheus") and f
.endswith(".j2")
6144 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6148 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6149 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6151 vnfr_id
= vnfr_id
.replace("-", "")
6153 "JOB_NAME": vnfr_id
,
6154 "TARGET_IP": target_ip
,
6155 "EXPORTER_POD_IP": host_name
,
6156 "EXPORTER_POD_PORT": host_port
,
6158 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
6159 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6160 for job
in job_list
:
6162 not isinstance(job
.get("job_name"), str)
6163 or vnfr_id
not in job
["job_name"]
6165 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6166 job
["nsr_id"] = nsr_id
6167 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
6168 if await self
.prometheus
.update(job_dict
):
6169 return list(job_dict
.keys())
6171 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6173 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6175 :param: vim_account_id: VIM Account ID
6177 :return: (cloud_name, cloud_credential)
6179 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6180 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6182 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6184 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6186 :param: vim_account_id: VIM Account ID
6188 :return: (cloud_name, cloud_credential)
6190 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6191 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")