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, cluster_type
="juju"
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
338 :cluster_type: The cluster type (juju, k8s)
342 # self.logger.debug("_on_update_k8s_db(cluster_uuid={}, kdu_instance={}, filter={}"
343 # .format(cluster_uuid, kdu_instance, filter))
345 nsr_id
= filter.get("_id")
347 vca_status
= await self
.k8scluster_map
[cluster_type
].status_kdu(
348 cluster_uuid
=cluster_uuid
,
349 kdu_instance
=kdu_instance
,
351 complete_status
=True,
357 db_dict
["vcaStatus"] = {nsr_id
: vca_status
}
359 if cluster_type
in ("juju-bundle", "juju"):
360 # TODO -> this should be done in a more uniform way, I think in N2VC, in order to update the K8s VCA
361 # status in a similar way between Juju Bundles and Helm Charts on this side
362 await self
.k8sclusterjuju
.update_vca_status(
363 db_dict
["vcaStatus"],
369 f
"Obtained VCA status for cluster type '{cluster_type}': {vca_status}"
373 self
.update_db_2("nsrs", nsr_id
, db_dict
)
374 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
376 except Exception as e
:
377 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
380 def _parse_cloud_init(cloud_init_text
, additional_params
, vnfd_id
, vdu_id
):
382 env
= Environment(undefined
=StrictUndefined
)
383 template
= env
.from_string(cloud_init_text
)
384 return template
.render(additional_params
or {})
385 except UndefinedError
as e
:
387 "Variable {} at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
388 "file, must be provided in the instantiation parameters inside the "
389 "'additionalParamsForVnf/Vdu' block".format(e
, vnfd_id
, vdu_id
)
391 except (TemplateError
, TemplateNotFound
) as e
:
393 "Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".format(
398 def _get_vdu_cloud_init_content(self
, vdu
, vnfd
):
399 cloud_init_content
= cloud_init_file
= None
401 if vdu
.get("cloud-init-file"):
402 base_folder
= vnfd
["_admin"]["storage"]
403 cloud_init_file
= "{}/{}/cloud_init/{}".format(
404 base_folder
["folder"],
405 base_folder
["pkg-dir"],
406 vdu
["cloud-init-file"],
408 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
409 cloud_init_content
= ci_file
.read()
410 elif vdu
.get("cloud-init"):
411 cloud_init_content
= vdu
["cloud-init"]
413 return cloud_init_content
414 except FsException
as e
:
416 "Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".format(
417 vnfd
["id"], vdu
["id"], cloud_init_file
, e
421 def _get_vdu_additional_params(self
, db_vnfr
, vdu_id
):
423 (vdur
for vdur
in db_vnfr
.get("vdur") if vdu_id
== vdur
["vdu-id-ref"]),
426 additional_params
= vdur
.get("additionalParams")
427 return parse_yaml_strings(additional_params
)
429 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
431 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
432 :param vnfd: input vnfd
433 :param new_id: overrides vnf id if provided
434 :param additionalParams: Instantiation params for VNFs provided
435 :param nsrId: Id of the NSR
436 :return: copy of vnfd
438 vnfd_RO
= deepcopy(vnfd
)
439 # remove unused by RO configuration, monitoring, scaling and internal keys
440 vnfd_RO
.pop("_id", None)
441 vnfd_RO
.pop("_admin", None)
442 vnfd_RO
.pop("monitoring-param", None)
443 vnfd_RO
.pop("scaling-group-descriptor", None)
444 vnfd_RO
.pop("kdu", None)
445 vnfd_RO
.pop("k8s-cluster", None)
447 vnfd_RO
["id"] = new_id
449 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
450 for vdu
in get_iterable(vnfd_RO
, "vdu"):
451 vdu
.pop("cloud-init-file", None)
452 vdu
.pop("cloud-init", None)
456 def ip_profile_2_RO(ip_profile
):
457 RO_ip_profile
= deepcopy(ip_profile
)
458 if "dns-server" in RO_ip_profile
:
459 if isinstance(RO_ip_profile
["dns-server"], list):
460 RO_ip_profile
["dns-address"] = []
461 for ds
in RO_ip_profile
.pop("dns-server"):
462 RO_ip_profile
["dns-address"].append(ds
["address"])
464 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
465 if RO_ip_profile
.get("ip-version") == "ipv4":
466 RO_ip_profile
["ip-version"] = "IPv4"
467 if RO_ip_profile
.get("ip-version") == "ipv6":
468 RO_ip_profile
["ip-version"] = "IPv6"
469 if "dhcp-params" in RO_ip_profile
:
470 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
473 def _get_ro_vim_id_for_vim_account(self
, vim_account
):
474 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
475 if db_vim
["_admin"]["operationalState"] != "ENABLED":
477 "VIM={} is not available. operationalState={}".format(
478 vim_account
, db_vim
["_admin"]["operationalState"]
481 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
484 def get_ro_wim_id_for_wim_account(self
, wim_account
):
485 if isinstance(wim_account
, str):
486 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
487 if db_wim
["_admin"]["operationalState"] != "ENABLED":
489 "WIM={} is not available. operationalState={}".format(
490 wim_account
, db_wim
["_admin"]["operationalState"]
493 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
498 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None, mark_delete
=False):
500 db_vdu_push_list
= []
502 db_update
= {"_admin.modified": time()}
504 for vdu_id
, vdu_count
in vdu_create
.items():
508 for vdur
in reversed(db_vnfr
["vdur"])
509 if vdur
["vdu-id-ref"] == vdu_id
514 # Read the template saved in the db:
515 self
.logger
.debug(f
"No vdur in the database. Using the vdur-template to scale")
516 vdur_template
= db_vnfr
.get("vdur-template")
517 if not vdur_template
:
519 "Error scaling OUT VNFR for {}. No vnfr or template exists".format(
523 vdur
= vdur_template
[0]
524 #Delete a template from the database after using it
525 self
.db
.set_one("vnfrs",
526 {"_id": db_vnfr
["_id"]},
528 pull
={"vdur-template": {"_id": vdur
['_id']}}
530 for count
in range(vdu_count
):
531 vdur_copy
= deepcopy(vdur
)
532 vdur_copy
["status"] = "BUILD"
533 vdur_copy
["status-detailed"] = None
534 vdur_copy
["ip-address"] = None
535 vdur_copy
["_id"] = str(uuid4())
536 vdur_copy
["count-index"] += count
+ 1
537 vdur_copy
["id"] = "{}-{}".format(
538 vdur_copy
["vdu-id-ref"], vdur_copy
["count-index"]
540 vdur_copy
.pop("vim_info", None)
541 for iface
in vdur_copy
["interfaces"]:
542 if iface
.get("fixed-ip"):
543 iface
["ip-address"] = self
.increment_ip_mac(
544 iface
["ip-address"], count
+ 1
547 iface
.pop("ip-address", None)
548 if iface
.get("fixed-mac"):
549 iface
["mac-address"] = self
.increment_ip_mac(
550 iface
["mac-address"], count
+ 1
553 iface
.pop("mac-address", None)
557 ) # only first vdu can be managment of vnf
558 db_vdu_push_list
.append(vdur_copy
)
559 # self.logger.debug("scale out, adding vdu={}".format(vdur_copy))
561 if len(db_vnfr
["vdur"]) == 1:
562 # The scale will move to 0 instances
563 self
.logger
.debug(f
"Scaling to 0 !, creating the template with the last vdur")
564 template_vdur
= [db_vnfr
["vdur"][0]]
565 for vdu_id
, vdu_count
in vdu_delete
.items():
567 indexes_to_delete
= [
569 for iv
in enumerate(db_vnfr
["vdur"])
570 if iv
[1]["vdu-id-ref"] == vdu_id
574 "vdur.{}.status".format(i
): "DELETING"
575 for i
in indexes_to_delete
[-vdu_count
:]
579 # it must be deleted one by one because common.db does not allow otherwise
582 for v
in reversed(db_vnfr
["vdur"])
583 if v
["vdu-id-ref"] == vdu_id
585 for vdu
in vdus_to_delete
[:vdu_count
]:
588 {"_id": db_vnfr
["_id"]},
590 pull
={"vdur": {"_id": vdu
["_id"]}},
594 db_push
["vdur"] = db_vdu_push_list
596 db_push
["vdur-template"] = template_vdur
599 db_vnfr
["vdur-template"] = template_vdur
600 self
.db
.set_one("vnfrs", {"_id": db_vnfr
["_id"]}, db_update
, push_list
=db_push
)
601 # modify passed dictionary db_vnfr
602 db_vnfr_
= self
.db
.get_one("vnfrs", {"_id": db_vnfr
["_id"]})
603 db_vnfr
["vdur"] = db_vnfr_
["vdur"]
605 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
607 Updates database nsr with the RO info for the created vld
608 :param ns_update_nsr: dictionary to be filled with the updated info
609 :param db_nsr: content of db_nsr. This is also modified
610 :param nsr_desc_RO: nsr descriptor from RO
611 :return: Nothing, LcmException is raised on errors
614 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
615 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
616 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
618 vld
["vim-id"] = net_RO
.get("vim_net_id")
619 vld
["name"] = net_RO
.get("vim_name")
620 vld
["status"] = net_RO
.get("status")
621 vld
["status-detailed"] = net_RO
.get("error_msg")
622 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
626 "ns_update_nsr: Not found vld={} at RO info".format(vld
["id"])
629 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
631 for db_vnfr
in db_vnfrs
.values():
632 vnfr_update
= {"status": "ERROR"}
633 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
634 if "status" not in vdur
:
635 vdur
["status"] = "ERROR"
636 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
638 vdur
["status-detailed"] = str(error_text
)
640 "vdur.{}.status-detailed".format(vdu_index
)
642 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
643 except DbException
as e
:
644 self
.logger
.error("Cannot update vnf. {}".format(e
))
646 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
648 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
649 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
650 :param nsr_desc_RO: nsr descriptor from RO
651 :return: Nothing, LcmException is raised on errors
653 for vnf_index
, db_vnfr
in db_vnfrs
.items():
654 for vnf_RO
in nsr_desc_RO
["vnfs"]:
655 if vnf_RO
["member_vnf_index"] != vnf_index
:
658 if vnf_RO
.get("ip_address"):
659 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
[
662 elif not db_vnfr
.get("ip-address"):
663 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
664 raise LcmExceptionNoMgmtIP(
665 "ns member_vnf_index '{}' has no IP address".format(
670 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
671 vdur_RO_count_index
= 0
672 if vdur
.get("pdu-type"):
674 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
675 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
677 if vdur
["count-index"] != vdur_RO_count_index
:
678 vdur_RO_count_index
+= 1
680 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
681 if vdur_RO
.get("ip_address"):
682 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
684 vdur
["ip-address"] = None
685 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
686 vdur
["name"] = vdur_RO
.get("vim_name")
687 vdur
["status"] = vdur_RO
.get("status")
688 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
689 for ifacer
in get_iterable(vdur
, "interfaces"):
690 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
691 if ifacer
["name"] == interface_RO
.get("internal_name"):
692 ifacer
["ip-address"] = interface_RO
.get(
695 ifacer
["mac-address"] = interface_RO
.get(
701 "ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
702 "from VIM info".format(
703 vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]
706 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
710 "ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
712 vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]
716 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
717 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
718 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
720 vld
["vim-id"] = net_RO
.get("vim_net_id")
721 vld
["name"] = net_RO
.get("vim_name")
722 vld
["status"] = net_RO
.get("status")
723 vld
["status-detailed"] = net_RO
.get("error_msg")
724 vnfr_update
["vld.{}".format(vld_index
)] = vld
728 "ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
733 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
738 "ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(
743 def _get_ns_config_info(self
, nsr_id
):
745 Generates a mapping between vnf,vdu elements and the N2VC id
746 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
747 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
748 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
749 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
751 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
752 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
754 ns_config_info
= {"osm-config-mapping": mapping
}
755 for vca
in vca_deployed_list
:
756 if not vca
["member-vnf-index"]:
758 if not vca
["vdu_id"]:
759 mapping
[vca
["member-vnf-index"]] = vca
["application"]
763 vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"]
765 ] = vca
["application"]
766 return ns_config_info
768 async def _instantiate_ng_ro(
785 def get_vim_account(vim_account_id
):
787 if vim_account_id
in db_vims
:
788 return db_vims
[vim_account_id
]
789 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account_id
})
790 db_vims
[vim_account_id
] = db_vim
793 # modify target_vld info with instantiation parameters
794 def parse_vld_instantiation_params(
795 target_vim
, target_vld
, vld_params
, target_sdn
797 if vld_params
.get("ip-profile"):
798 target_vld
["vim_info"][target_vim
]["ip_profile"] = vld_params
[
801 if vld_params
.get("provider-network"):
802 target_vld
["vim_info"][target_vim
]["provider_network"] = vld_params
[
805 if "sdn-ports" in vld_params
["provider-network"] and target_sdn
:
806 target_vld
["vim_info"][target_sdn
]["sdn-ports"] = vld_params
[
809 if vld_params
.get("wimAccountId"):
810 target_wim
= "wim:{}".format(vld_params
["wimAccountId"])
811 target_vld
["vim_info"][target_wim
] = {}
812 for param
in ("vim-network-name", "vim-network-id"):
813 if vld_params
.get(param
):
814 if isinstance(vld_params
[param
], dict):
815 for vim
, vim_net
in vld_params
[param
].items():
816 other_target_vim
= "vim:" + vim
818 target_vld
["vim_info"],
819 (other_target_vim
, param
.replace("-", "_")),
822 else: # isinstance str
823 target_vld
["vim_info"][target_vim
][
824 param
.replace("-", "_")
825 ] = vld_params
[param
]
826 if vld_params
.get("common_id"):
827 target_vld
["common_id"] = vld_params
.get("common_id")
829 # modify target["ns"]["vld"] with instantiation parameters to override vnf vim-account
830 def update_ns_vld_target(target
, ns_params
):
831 for vnf_params
in ns_params
.get("vnf", ()):
832 if vnf_params
.get("vimAccountId"):
836 for vnfr
in db_vnfrs
.values()
837 if vnf_params
["member-vnf-index"]
838 == vnfr
["member-vnf-index-ref"]
842 vdur
= next((vdur
for vdur
in target_vnf
.get("vdur", ())), None)
843 for a_index
, a_vld
in enumerate(target
["ns"]["vld"]):
844 target_vld
= find_in_list(
845 get_iterable(vdur
, "interfaces"),
846 lambda iface
: iface
.get("ns-vld-id") == a_vld
["name"],
849 if vnf_params
.get("vimAccountId") not in a_vld
.get(
852 target
["ns"]["vld"][a_index
].get("vim_info").update(
854 "vim:{}".format(vnf_params
["vimAccountId"]): {
855 "vim_network_name": ""
860 nslcmop_id
= db_nslcmop
["_id"]
862 "name": db_nsr
["name"],
865 "image": deepcopy(db_nsr
["image"]),
866 "flavor": deepcopy(db_nsr
["flavor"]),
867 "action_id": nslcmop_id
,
868 "cloud_init_content": {},
870 for image
in target
["image"]:
871 image
["vim_info"] = {}
872 for flavor
in target
["flavor"]:
873 flavor
["vim_info"] = {}
874 if db_nsr
.get("affinity-or-anti-affinity-group"):
875 target
["affinity-or-anti-affinity-group"] = deepcopy(db_nsr
["affinity-or-anti-affinity-group"])
876 for affinity_or_anti_affinity_group
in target
["affinity-or-anti-affinity-group"]:
877 affinity_or_anti_affinity_group
["vim_info"] = {}
879 if db_nslcmop
.get("lcmOperationType") != "instantiate":
880 # get parameters of instantiation:
881 db_nslcmop_instantiate
= self
.db
.get_list(
884 "nsInstanceId": db_nslcmop
["nsInstanceId"],
885 "lcmOperationType": "instantiate",
888 ns_params
= db_nslcmop_instantiate
.get("operationParams")
890 ns_params
= db_nslcmop
.get("operationParams")
891 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
892 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
895 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
896 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
900 "mgmt-network": vld
.get("mgmt-network", False),
901 "type": vld
.get("type"),
904 "vim_network_name": vld
.get("vim-network-name"),
905 "vim_account_id": ns_params
["vimAccountId"],
909 # check if this network needs SDN assist
910 if vld
.get("pci-interfaces"):
911 db_vim
= get_vim_account(ns_params
["vimAccountId"])
912 sdnc_id
= db_vim
["config"].get("sdn-controller")
914 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
915 target_sdn
= "sdn:{}".format(sdnc_id
)
916 target_vld
["vim_info"][target_sdn
] = {
918 "target_vim": target_vim
,
920 "type": vld
.get("type"),
923 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
924 for nsd_vnf_profile
in nsd_vnf_profiles
:
925 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
926 if cp
["virtual-link-profile-id"] == vld
["id"]:
928 "member_vnf:{}.{}".format(
929 cp
["constituent-cpd-id"][0][
930 "constituent-base-element-id"
932 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
934 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
936 # check at nsd descriptor, if there is an ip-profile
938 nsd_vlp
= find_in_list(
939 get_virtual_link_profiles(nsd
),
940 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
945 and nsd_vlp
.get("virtual-link-protocol-data")
946 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
948 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
951 ip_profile_dest_data
= {}
952 if "ip-version" in ip_profile_source_data
:
953 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
956 if "cidr" in ip_profile_source_data
:
957 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
960 if "gateway-ip" in ip_profile_source_data
:
961 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
964 if "dhcp-enabled" in ip_profile_source_data
:
965 ip_profile_dest_data
["dhcp-params"] = {
966 "enabled": ip_profile_source_data
["dhcp-enabled"]
968 vld_params
["ip-profile"] = ip_profile_dest_data
970 # update vld_params with instantiation params
971 vld_instantiation_params
= find_in_list(
972 get_iterable(ns_params
, "vld"),
973 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
975 if vld_instantiation_params
:
976 vld_params
.update(vld_instantiation_params
)
977 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
978 target
["ns"]["vld"].append(target_vld
)
979 # Update the target ns_vld if vnf vim_account is overriden by instantiation params
980 update_ns_vld_target(target
, ns_params
)
982 for vnfr
in db_vnfrs
.values():
984 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
986 vnf_params
= find_in_list(
987 get_iterable(ns_params
, "vnf"),
988 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
990 target_vnf
= deepcopy(vnfr
)
991 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
992 for vld
in target_vnf
.get("vld", ()):
993 # check if connected to a ns.vld, to fill target'
994 vnf_cp
= find_in_list(
995 vnfd
.get("int-virtual-link-desc", ()),
996 lambda cpd
: cpd
.get("id") == vld
["id"],
999 ns_cp
= "member_vnf:{}.{}".format(
1000 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
1002 if cp2target
.get(ns_cp
):
1003 vld
["target"] = cp2target
[ns_cp
]
1006 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
1008 # check if this network needs SDN assist
1010 if vld
.get("pci-interfaces"):
1011 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1012 sdnc_id
= db_vim
["config"].get("sdn-controller")
1014 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
1015 target_sdn
= "sdn:{}".format(sdnc_id
)
1016 vld
["vim_info"][target_sdn
] = {
1018 "target_vim": target_vim
,
1020 "type": vld
.get("type"),
1023 # check at vnfd descriptor, if there is an ip-profile
1025 vnfd_vlp
= find_in_list(
1026 get_virtual_link_profiles(vnfd
),
1027 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
1031 and vnfd_vlp
.get("virtual-link-protocol-data")
1032 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
1034 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
1037 ip_profile_dest_data
= {}
1038 if "ip-version" in ip_profile_source_data
:
1039 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
1042 if "cidr" in ip_profile_source_data
:
1043 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
1046 if "gateway-ip" in ip_profile_source_data
:
1047 ip_profile_dest_data
[
1049 ] = ip_profile_source_data
["gateway-ip"]
1050 if "dhcp-enabled" in ip_profile_source_data
:
1051 ip_profile_dest_data
["dhcp-params"] = {
1052 "enabled": ip_profile_source_data
["dhcp-enabled"]
1055 vld_params
["ip-profile"] = ip_profile_dest_data
1056 # update vld_params with instantiation params
1058 vld_instantiation_params
= find_in_list(
1059 get_iterable(vnf_params
, "internal-vld"),
1060 lambda i_vld
: i_vld
["name"] == vld
["id"],
1062 if vld_instantiation_params
:
1063 vld_params
.update(vld_instantiation_params
)
1064 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1067 for vdur
in target_vnf
.get("vdur", ()):
1068 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1069 continue # This vdu must not be created
1070 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1072 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1075 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1076 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1079 and vdu_configuration
.get("config-access")
1080 and vdu_configuration
.get("config-access").get("ssh-access")
1082 vdur
["ssh-keys"] = ssh_keys_all
1083 vdur
["ssh-access-required"] = vdu_configuration
[
1085 ]["ssh-access"]["required"]
1088 and vnf_configuration
.get("config-access")
1089 and vnf_configuration
.get("config-access").get("ssh-access")
1090 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1092 vdur
["ssh-keys"] = ssh_keys_all
1093 vdur
["ssh-access-required"] = vnf_configuration
[
1095 ]["ssh-access"]["required"]
1096 elif ssh_keys_instantiation
and find_in_list(
1097 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1099 vdur
["ssh-keys"] = ssh_keys_instantiation
1101 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1103 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1105 if vdud
.get("cloud-init-file"):
1106 vdur
["cloud-init"] = "{}:file:{}".format(
1107 vnfd
["_id"], vdud
.get("cloud-init-file")
1109 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1110 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1111 base_folder
= vnfd
["_admin"]["storage"]
1112 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1113 base_folder
["folder"],
1114 base_folder
["pkg-dir"],
1115 vdud
.get("cloud-init-file"),
1117 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
1118 target
["cloud_init_content"][
1121 elif vdud
.get("cloud-init"):
1122 vdur
["cloud-init"] = "{}:vdu:{}".format(
1123 vnfd
["_id"], get_vdu_index(vnfd
, vdur
["vdu-id-ref"])
1125 # put content at target.cloul_init_content. Avoid ng_ro read vnfd descriptor
1126 target
["cloud_init_content"][vdur
["cloud-init"]] = vdud
[
1129 vdur
["additionalParams"] = vdur
.get("additionalParams") or {}
1130 deploy_params_vdu
= self
._format
_additional
_params
(
1131 vdur
.get("additionalParams") or {}
1133 deploy_params_vdu
["OSM"] = get_osm_params(
1134 vnfr
, vdur
["vdu-id-ref"], vdur
["count-index"]
1136 vdur
["additionalParams"] = deploy_params_vdu
1139 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
1140 if target_vim
not in ns_flavor
["vim_info"]:
1141 ns_flavor
["vim_info"][target_vim
] = {}
1144 # in case alternative images are provided we must check if they should be applied
1145 # for the vim_type, modify the vim_type taking into account
1146 ns_image_id
= int(vdur
["ns-image-id"])
1147 if vdur
.get("alt-image-ids"):
1148 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1149 vim_type
= db_vim
["vim_type"]
1150 for alt_image_id
in vdur
.get("alt-image-ids"):
1151 ns_alt_image
= target
["image"][int(alt_image_id
)]
1152 if vim_type
== ns_alt_image
.get("vim-type"):
1153 # must use alternative image
1155 "use alternative image id: {}".format(alt_image_id
)
1157 ns_image_id
= alt_image_id
1158 vdur
["ns-image-id"] = ns_image_id
1160 ns_image
= target
["image"][int(ns_image_id
)]
1161 if target_vim
not in ns_image
["vim_info"]:
1162 ns_image
["vim_info"][target_vim
] = {}
1165 if vdur
.get("affinity-or-anti-affinity-group-id"):
1166 for ags_id
in vdur
["affinity-or-anti-affinity-group-id"]:
1167 ns_ags
= target
["affinity-or-anti-affinity-group"][int(ags_id
)]
1168 if target_vim
not in ns_ags
["vim_info"]:
1169 ns_ags
["vim_info"][target_vim
] = {}
1171 vdur
["vim_info"] = {target_vim
: {}}
1172 # instantiation parameters
1174 # vdu_instantiation_params = next((v for v in get_iterable(vnf_params, "vdu") if v["id"] ==
1175 # vdud["id"]), None)
1176 vdur_list
.append(vdur
)
1177 target_vnf
["vdur"] = vdur_list
1178 target
["vnf"].append(target_vnf
)
1180 desc
= await self
.RO
.deploy(nsr_id
, target
)
1181 self
.logger
.debug("RO return > {}".format(desc
))
1182 action_id
= desc
["action_id"]
1183 await self
._wait
_ng
_ro
(
1184 nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
1189 "_admin.deployed.RO.operational-status": "running",
1190 "detailed-status": " ".join(stage
),
1192 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1193 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1194 self
._write
_op
_status
(nslcmop_id
, stage
)
1196 logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
)
1200 async def _wait_ng_ro(
1209 detailed_status_old
= None
1211 start_time
= start_time
or time()
1212 while time() <= start_time
+ timeout
:
1213 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
1214 self
.logger
.debug("Wait NG RO > {}".format(desc_status
))
1215 if desc_status
["status"] == "FAILED":
1216 raise NgRoException(desc_status
["details"])
1217 elif desc_status
["status"] == "BUILD":
1219 stage
[2] = "VIM: ({})".format(desc_status
["details"])
1220 elif desc_status
["status"] == "DONE":
1222 stage
[2] = "Deployed at VIM"
1225 assert False, "ROclient.check_ns_status returns unknown {}".format(
1226 desc_status
["status"]
1228 if stage
and nslcmop_id
and stage
[2] != detailed_status_old
:
1229 detailed_status_old
= stage
[2]
1230 db_nsr_update
["detailed-status"] = " ".join(stage
)
1231 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1232 self
._write
_op
_status
(nslcmop_id
, stage
)
1233 await asyncio
.sleep(15, loop
=self
.loop
)
1234 else: # timeout_ns_deploy
1235 raise NgRoException("Timeout waiting ns to deploy")
1237 async def _terminate_ng_ro(
1238 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
1243 start_deploy
= time()
1250 "action_id": nslcmop_id
,
1252 desc
= await self
.RO
.deploy(nsr_id
, target
)
1253 action_id
= desc
["action_id"]
1254 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1255 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1258 + "ns terminate action at RO. action_id={}".format(action_id
)
1262 delete_timeout
= 20 * 60 # 20 minutes
1263 await self
._wait
_ng
_ro
(
1264 nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
1267 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1268 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1270 await self
.RO
.delete(nsr_id
)
1271 except Exception as e
:
1272 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1273 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1274 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1275 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1277 logging_text
+ "RO_action_id={} already deleted".format(action_id
)
1279 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1280 failed_detail
.append("delete conflict: {}".format(e
))
1283 + "RO_action_id={} delete conflict: {}".format(action_id
, e
)
1286 failed_detail
.append("delete error: {}".format(e
))
1289 + "RO_action_id={} delete error: {}".format(action_id
, e
)
1293 stage
[2] = "Error deleting from VIM"
1295 stage
[2] = "Deleted from VIM"
1296 db_nsr_update
["detailed-status"] = " ".join(stage
)
1297 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1298 self
._write
_op
_status
(nslcmop_id
, stage
)
1301 raise LcmException("; ".join(failed_detail
))
1304 async def instantiate_RO(
1318 :param logging_text: preffix text to use at logging
1319 :param nsr_id: nsr identity
1320 :param nsd: database content of ns descriptor
1321 :param db_nsr: database content of ns record
1322 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1324 :param db_vnfds: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1325 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1326 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1327 :return: None or exception
1330 start_deploy
= time()
1331 ns_params
= db_nslcmop
.get("operationParams")
1332 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1333 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1335 timeout_ns_deploy
= self
.timeout
.get(
1336 "ns_deploy", self
.timeout_ns_deploy
1339 # Check for and optionally request placement optimization. Database will be updated if placement activated
1340 stage
[2] = "Waiting for Placement."
1341 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1342 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1343 for vnfr
in db_vnfrs
.values():
1344 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1347 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1349 return await self
._instantiate
_ng
_ro
(
1362 except Exception as e
:
1363 stage
[2] = "ERROR deploying at VIM"
1364 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1366 "Error deploying at VIM {}".format(e
),
1367 exc_info
=not isinstance(
1370 ROclient
.ROClientException
,
1379 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1381 Wait for kdu to be up, get ip address
1382 :param logging_text: prefix use for logging
1389 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1392 while nb_tries
< 360:
1393 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1397 for x
in get_iterable(db_vnfr
, "kdur")
1398 if x
.get("kdu-name") == kdu_name
1404 "Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
)
1406 if kdur
.get("status"):
1407 if kdur
["status"] in ("READY", "ENABLED"):
1408 return kdur
.get("ip-address")
1411 "target KDU={} is in error state".format(kdu_name
)
1414 await asyncio
.sleep(10, loop
=self
.loop
)
1416 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1418 async def wait_vm_up_insert_key_ro(
1419 self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None
1422 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1423 :param logging_text: prefix use for logging
1428 :param pub_key: public ssh key to inject, None to skip
1429 :param user: user to apply the public ssh key
1433 self
.logger
.debug(logging_text
+ "Starting wait_vm_up_insert_key_ro")
1437 target_vdu_id
= None
1443 if ro_retries
>= 360: # 1 hour
1445 "Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
)
1448 await asyncio
.sleep(10, loop
=self
.loop
)
1451 if not target_vdu_id
:
1452 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1454 if not vdu_id
: # for the VNF case
1455 if db_vnfr
.get("status") == "ERROR":
1457 "Cannot inject ssh-key because target VNF is in error state"
1459 ip_address
= db_vnfr
.get("ip-address")
1465 for x
in get_iterable(db_vnfr
, "vdur")
1466 if x
.get("ip-address") == ip_address
1474 for x
in get_iterable(db_vnfr
, "vdur")
1475 if x
.get("vdu-id-ref") == vdu_id
1476 and x
.get("count-index") == vdu_index
1482 not vdur
and len(db_vnfr
.get("vdur", ())) == 1
1483 ): # If only one, this should be the target vdu
1484 vdur
= db_vnfr
["vdur"][0]
1487 "Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(
1488 vnfr_id
, vdu_id
, vdu_index
1491 # New generation RO stores information at "vim_info"
1494 if vdur
.get("vim_info"):
1496 t
for t
in vdur
["vim_info"]
1497 ) # there should be only one key
1498 ng_ro_status
= vdur
["vim_info"][target_vim
].get("vim_status")
1500 vdur
.get("pdu-type")
1501 or vdur
.get("status") == "ACTIVE"
1502 or ng_ro_status
== "ACTIVE"
1504 ip_address
= vdur
.get("ip-address")
1507 target_vdu_id
= vdur
["vdu-id-ref"]
1508 elif vdur
.get("status") == "ERROR" or ng_ro_status
== "ERROR":
1510 "Cannot inject ssh-key because target VM is in error state"
1513 if not target_vdu_id
:
1516 # inject public key into machine
1517 if pub_key
and user
:
1518 self
.logger
.debug(logging_text
+ "Inserting RO key")
1519 self
.logger
.debug("SSH > PubKey > {}".format(pub_key
))
1520 if vdur
.get("pdu-type"):
1521 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1524 ro_vm_id
= "{}-{}".format(
1525 db_vnfr
["member-vnf-index-ref"], target_vdu_id
1526 ) # TODO add vdu_index
1530 "action": "inject_ssh_key",
1534 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdur
["id"]}]}],
1536 desc
= await self
.RO
.deploy(nsr_id
, target
)
1537 action_id
= desc
["action_id"]
1538 await self
._wait
_ng
_ro
(nsr_id
, action_id
, timeout
=600)
1541 # wait until NS is deployed at RO
1543 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1544 ro_nsr_id
= deep_get(
1545 db_nsrs
, ("_admin", "deployed", "RO", "nsr_id")
1549 result_dict
= await self
.RO
.create_action(
1551 item_id_name
=ro_nsr_id
,
1553 "add_public_key": pub_key
,
1558 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1559 if not result_dict
or not isinstance(result_dict
, dict):
1561 "Unknown response from RO when injecting key"
1563 for result
in result_dict
.values():
1564 if result
.get("vim_result") == 200:
1567 raise ROclient
.ROClientException(
1568 "error injecting key: {}".format(
1569 result
.get("description")
1573 except NgRoException
as e
:
1575 "Reaching max tries injecting key. Error: {}".format(e
)
1577 except ROclient
.ROClientException
as e
:
1581 + "error injecting key: {}. Retrying until {} seconds".format(
1588 "Reaching max tries injecting key. Error: {}".format(e
)
1595 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1597 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1599 my_vca
= vca_deployed_list
[vca_index
]
1600 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1601 # vdu or kdu: no dependencies
1605 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1606 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1607 configuration_status_list
= db_nsr
["configurationStatus"]
1608 for index
, vca_deployed
in enumerate(configuration_status_list
):
1609 if index
== vca_index
:
1612 if not my_vca
.get("member-vnf-index") or (
1613 vca_deployed
.get("member-vnf-index")
1614 == my_vca
.get("member-vnf-index")
1616 internal_status
= configuration_status_list
[index
].get("status")
1617 if internal_status
== "READY":
1619 elif internal_status
== "BROKEN":
1621 "Configuration aborted because dependent charm/s has failed"
1626 # no dependencies, return
1628 await asyncio
.sleep(10)
1631 raise LcmException("Configuration aborted because dependent charm/s timeout")
1633 def get_vca_id(self
, db_vnfr
: dict, db_nsr
: dict):
1636 vca_id
= deep_get(db_vnfr
, ("vca-id",))
1638 vim_account_id
= deep_get(db_nsr
, ("instantiate_params", "vimAccountId"))
1639 vca_id
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("vca")
1642 async def instantiate_N2VC(
1659 ee_config_descriptor
,
1661 nsr_id
= db_nsr
["_id"]
1662 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1663 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1664 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1665 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1667 "collection": "nsrs",
1668 "filter": {"_id": nsr_id
},
1669 "path": db_update_entry
,
1675 element_under_configuration
= nsr_id
1679 vnfr_id
= db_vnfr
["_id"]
1680 osm_config
["osm"]["vnf_id"] = vnfr_id
1682 namespace
= "{nsi}.{ns}".format(nsi
=nsi_id
if nsi_id
else "", ns
=nsr_id
)
1684 if vca_type
== "native_charm":
1687 index_number
= vdu_index
or 0
1690 element_type
= "VNF"
1691 element_under_configuration
= vnfr_id
1692 namespace
+= ".{}-{}".format(vnfr_id
, index_number
)
1694 namespace
+= ".{}-{}".format(vdu_id
, index_number
)
1695 element_type
= "VDU"
1696 element_under_configuration
= "{}-{}".format(vdu_id
, index_number
)
1697 osm_config
["osm"]["vdu_id"] = vdu_id
1699 namespace
+= ".{}".format(kdu_name
)
1700 element_type
= "KDU"
1701 element_under_configuration
= kdu_name
1702 osm_config
["osm"]["kdu_name"] = kdu_name
1705 artifact_path
= "{}/{}/{}/{}".format(
1706 base_folder
["folder"],
1707 base_folder
["pkg-dir"],
1709 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1714 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1716 # get initial_config_primitive_list that applies to this element
1717 initial_config_primitive_list
= config_descriptor
.get(
1718 "initial-config-primitive"
1722 "Initial config primitive list > {}".format(
1723 initial_config_primitive_list
1727 # add config if not present for NS charm
1728 ee_descriptor_id
= ee_config_descriptor
.get("id")
1729 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1730 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1731 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1735 "Initial config primitive list #2 > {}".format(
1736 initial_config_primitive_list
1739 # n2vc_redesign STEP 3.1
1740 # find old ee_id if exists
1741 ee_id
= vca_deployed
.get("ee_id")
1743 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1744 # create or register execution environment in VCA
1745 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1747 self
._write
_configuration
_status
(
1749 vca_index
=vca_index
,
1751 element_under_configuration
=element_under_configuration
,
1752 element_type
=element_type
,
1755 step
= "create execution environment"
1756 self
.logger
.debug(logging_text
+ step
)
1760 if vca_type
== "k8s_proxy_charm":
1761 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1762 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1763 namespace
=namespace
,
1764 artifact_path
=artifact_path
,
1768 elif vca_type
== "helm" or vca_type
== "helm-v3":
1769 ee_id
, credentials
= await self
.vca_map
[
1771 ].create_execution_environment(
1772 namespace
=namespace
,
1776 artifact_path
=artifact_path
,
1780 ee_id
, credentials
= await self
.vca_map
[
1782 ].create_execution_environment(
1783 namespace
=namespace
,
1789 elif vca_type
== "native_charm":
1790 step
= "Waiting to VM being up and getting IP address"
1791 self
.logger
.debug(logging_text
+ step
)
1792 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1801 credentials
= {"hostname": rw_mgmt_ip
}
1803 username
= deep_get(
1804 config_descriptor
, ("config-access", "ssh-access", "default-user")
1806 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1807 # merged. Meanwhile let's get username from initial-config-primitive
1808 if not username
and initial_config_primitive_list
:
1809 for config_primitive
in initial_config_primitive_list
:
1810 for param
in config_primitive
.get("parameter", ()):
1811 if param
["name"] == "ssh-username":
1812 username
= param
["value"]
1816 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1817 "'config-access.ssh-access.default-user'"
1819 credentials
["username"] = username
1820 # n2vc_redesign STEP 3.2
1822 self
._write
_configuration
_status
(
1824 vca_index
=vca_index
,
1825 status
="REGISTERING",
1826 element_under_configuration
=element_under_configuration
,
1827 element_type
=element_type
,
1830 step
= "register execution environment {}".format(credentials
)
1831 self
.logger
.debug(logging_text
+ step
)
1832 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1833 credentials
=credentials
,
1834 namespace
=namespace
,
1839 # for compatibility with MON/POL modules, the need model and application name at database
1840 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1841 ee_id_parts
= ee_id
.split(".")
1842 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1843 if len(ee_id_parts
) >= 2:
1844 model_name
= ee_id_parts
[0]
1845 application_name
= ee_id_parts
[1]
1846 db_nsr_update
[db_update_entry
+ "model"] = model_name
1847 db_nsr_update
[db_update_entry
+ "application"] = application_name
1849 # n2vc_redesign STEP 3.3
1850 step
= "Install configuration Software"
1852 self
._write
_configuration
_status
(
1854 vca_index
=vca_index
,
1855 status
="INSTALLING SW",
1856 element_under_configuration
=element_under_configuration
,
1857 element_type
=element_type
,
1858 other_update
=db_nsr_update
,
1861 # TODO check if already done
1862 self
.logger
.debug(logging_text
+ step
)
1864 if vca_type
== "native_charm":
1865 config_primitive
= next(
1866 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1869 if config_primitive
:
1870 config
= self
._map
_primitive
_params
(
1871 config_primitive
, {}, deploy_params
1874 if vca_type
== "lxc_proxy_charm":
1875 if element_type
== "NS":
1876 num_units
= db_nsr
.get("config-units") or 1
1877 elif element_type
== "VNF":
1878 num_units
= db_vnfr
.get("config-units") or 1
1879 elif element_type
== "VDU":
1880 for v
in db_vnfr
["vdur"]:
1881 if vdu_id
== v
["vdu-id-ref"]:
1882 num_units
= v
.get("config-units") or 1
1884 if vca_type
!= "k8s_proxy_charm":
1885 await self
.vca_map
[vca_type
].install_configuration_sw(
1887 artifact_path
=artifact_path
,
1890 num_units
=num_units
,
1895 # write in db flag of configuration_sw already installed
1897 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1900 # add relations for this VCA (wait for other peers related with this VCA)
1901 await self
._add
_vca
_relations
(
1902 logging_text
=logging_text
,
1904 vca_index
=vca_index
,
1909 # if SSH access is required, then get execution environment SSH public
1910 # if native charm we have waited already to VM be UP
1911 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1914 # self.logger.debug("get ssh key block")
1916 config_descriptor
, ("config-access", "ssh-access", "required")
1918 # self.logger.debug("ssh key needed")
1919 # Needed to inject a ssh key
1922 ("config-access", "ssh-access", "default-user"),
1924 step
= "Install configuration Software, getting public ssh key"
1925 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1926 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1929 step
= "Insert public key into VM user={} ssh_key={}".format(
1933 # self.logger.debug("no need to get ssh key")
1934 step
= "Waiting to VM being up and getting IP address"
1935 self
.logger
.debug(logging_text
+ step
)
1937 # n2vc_redesign STEP 5.1
1938 # wait for RO (ip-address) Insert pub_key into VM
1941 rw_mgmt_ip
= await self
.wait_kdu_up(
1942 logging_text
, nsr_id
, vnfr_id
, kdu_name
1945 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1955 rw_mgmt_ip
= None # This is for a NS configuration
1957 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1959 # store rw_mgmt_ip in deploy params for later replacement
1960 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1962 # n2vc_redesign STEP 6 Execute initial config primitive
1963 step
= "execute initial config primitive"
1965 # wait for dependent primitives execution (NS -> VNF -> VDU)
1966 if initial_config_primitive_list
:
1967 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1969 # stage, in function of element type: vdu, kdu, vnf or ns
1970 my_vca
= vca_deployed_list
[vca_index
]
1971 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1973 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1974 elif my_vca
.get("member-vnf-index"):
1976 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1979 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1981 self
._write
_configuration
_status
(
1982 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1985 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1987 check_if_terminated_needed
= True
1988 for initial_config_primitive
in initial_config_primitive_list
:
1989 # adding information on the vca_deployed if it is a NS execution environment
1990 if not vca_deployed
["member-vnf-index"]:
1991 deploy_params
["ns_config_info"] = json
.dumps(
1992 self
._get
_ns
_config
_info
(nsr_id
)
1994 # TODO check if already done
1995 primitive_params_
= self
._map
_primitive
_params
(
1996 initial_config_primitive
, {}, deploy_params
1999 step
= "execute primitive '{}' params '{}'".format(
2000 initial_config_primitive
["name"], primitive_params_
2002 self
.logger
.debug(logging_text
+ step
)
2003 await self
.vca_map
[vca_type
].exec_primitive(
2005 primitive_name
=initial_config_primitive
["name"],
2006 params_dict
=primitive_params_
,
2011 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
2012 if check_if_terminated_needed
:
2013 if config_descriptor
.get("terminate-config-primitive"):
2015 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
2017 check_if_terminated_needed
= False
2019 # TODO register in database that primitive is done
2021 # STEP 7 Configure metrics
2022 if vca_type
== "helm" or vca_type
== "helm-v3":
2023 prometheus_jobs
= await self
.add_prometheus_metrics(
2025 artifact_path
=artifact_path
,
2026 ee_config_descriptor
=ee_config_descriptor
,
2029 target_ip
=rw_mgmt_ip
,
2035 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
2038 step
= "instantiated at VCA"
2039 self
.logger
.debug(logging_text
+ step
)
2041 self
._write
_configuration
_status
(
2042 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
2045 except Exception as e
: # TODO not use Exception but N2VC exception
2046 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
2048 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
2051 "Exception while {} : {}".format(step
, e
), exc_info
=True
2053 self
._write
_configuration
_status
(
2054 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
2056 raise LcmException("{} {}".format(step
, e
)) from e
2058 def _write_ns_status(
2062 current_operation
: str,
2063 current_operation_id
: str,
2064 error_description
: str = None,
2065 error_detail
: str = None,
2066 other_update
: dict = None,
2069 Update db_nsr fields.
2072 :param current_operation:
2073 :param current_operation_id:
2074 :param error_description:
2075 :param error_detail:
2076 :param other_update: Other required changes at database if provided, will be cleared
2080 db_dict
= other_update
or {}
2083 ] = current_operation_id
# for backward compatibility
2084 db_dict
["_admin.current-operation"] = current_operation_id
2085 db_dict
["_admin.operation-type"] = (
2086 current_operation
if current_operation
!= "IDLE" else None
2088 db_dict
["currentOperation"] = current_operation
2089 db_dict
["currentOperationID"] = current_operation_id
2090 db_dict
["errorDescription"] = error_description
2091 db_dict
["errorDetail"] = error_detail
2094 db_dict
["nsState"] = ns_state
2095 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2096 except DbException
as e
:
2097 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2099 def _write_op_status(
2103 error_message
: str = None,
2104 queuePosition
: int = 0,
2105 operation_state
: str = None,
2106 other_update
: dict = None,
2109 db_dict
= other_update
or {}
2110 db_dict
["queuePosition"] = queuePosition
2111 if isinstance(stage
, list):
2112 db_dict
["stage"] = stage
[0]
2113 db_dict
["detailed-status"] = " ".join(stage
)
2114 elif stage
is not None:
2115 db_dict
["stage"] = str(stage
)
2117 if error_message
is not None:
2118 db_dict
["errorMessage"] = error_message
2119 if operation_state
is not None:
2120 db_dict
["operationState"] = operation_state
2121 db_dict
["statusEnteredTime"] = time()
2122 self
.update_db_2("nslcmops", op_id
, db_dict
)
2123 except DbException
as e
:
2125 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2128 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2130 nsr_id
= db_nsr
["_id"]
2131 # configurationStatus
2132 config_status
= db_nsr
.get("configurationStatus")
2135 "configurationStatus.{}.status".format(index
): status
2136 for index
, v
in enumerate(config_status
)
2140 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2142 except DbException
as e
:
2144 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2147 def _write_configuration_status(
2152 element_under_configuration
: str = None,
2153 element_type
: str = None,
2154 other_update
: dict = None,
2157 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2158 # .format(vca_index, status))
2161 db_path
= "configurationStatus.{}.".format(vca_index
)
2162 db_dict
= other_update
or {}
2164 db_dict
[db_path
+ "status"] = status
2165 if element_under_configuration
:
2167 db_path
+ "elementUnderConfiguration"
2168 ] = element_under_configuration
2170 db_dict
[db_path
+ "elementType"] = element_type
2171 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2172 except DbException
as e
:
2174 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2175 status
, nsr_id
, vca_index
, e
2179 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2181 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2182 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2183 Database is used because the result can be obtained from a different LCM worker in case of HA.
2184 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2185 :param db_nslcmop: database content of nslcmop
2186 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2187 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2188 computed 'vim-account-id'
2191 nslcmop_id
= db_nslcmop
["_id"]
2192 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2193 if placement_engine
== "PLA":
2195 logging_text
+ "Invoke and wait for placement optimization"
2197 await self
.msg
.aiowrite(
2198 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2200 db_poll_interval
= 5
2201 wait
= db_poll_interval
* 10
2203 while not pla_result
and wait
>= 0:
2204 await asyncio
.sleep(db_poll_interval
)
2205 wait
-= db_poll_interval
2206 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2207 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2211 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2214 for pla_vnf
in pla_result
["vnf"]:
2215 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2216 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2221 {"_id": vnfr
["_id"]},
2222 {"vim-account-id": pla_vnf
["vimAccountId"]},
2225 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2228 def update_nsrs_with_pla_result(self
, params
):
2230 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2232 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2234 except Exception as e
:
2235 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2237 async def instantiate(self
, nsr_id
, nslcmop_id
):
2240 :param nsr_id: ns instance to deploy
2241 :param nslcmop_id: operation to run
2245 # Try to lock HA task here
2246 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2247 if not task_is_locked_by_me
:
2249 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2253 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2254 self
.logger
.debug(logging_text
+ "Enter")
2256 # get all needed from database
2258 # database nsrs record
2261 # database nslcmops record
2264 # update operation on nsrs
2266 # update operation on nslcmops
2267 db_nslcmop_update
= {}
2269 nslcmop_operation_state
= None
2270 db_vnfrs
= {} # vnf's info indexed by member-index
2272 tasks_dict_info
= {} # from task to info text
2276 "Stage 1/5: preparation of the environment.",
2277 "Waiting for previous operations to terminate.",
2280 # ^ stage, step, VIM progress
2282 # wait for any previous tasks in process
2283 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2285 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2286 stage
[1] = "Reading from database."
2287 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2288 db_nsr_update
["detailed-status"] = "creating"
2289 db_nsr_update
["operational-status"] = "init"
2290 self
._write
_ns
_status
(
2292 ns_state
="BUILDING",
2293 current_operation
="INSTANTIATING",
2294 current_operation_id
=nslcmop_id
,
2295 other_update
=db_nsr_update
,
2297 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2299 # read from db: operation
2300 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2301 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2302 if db_nslcmop
["operationParams"].get("additionalParamsForVnf"):
2303 db_nslcmop
["operationParams"]["additionalParamsForVnf"] = json
.loads(
2304 db_nslcmop
["operationParams"]["additionalParamsForVnf"]
2306 ns_params
= db_nslcmop
.get("operationParams")
2307 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2308 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2310 timeout_ns_deploy
= self
.timeout
.get(
2311 "ns_deploy", self
.timeout_ns_deploy
2315 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2316 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2317 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2318 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2319 self
.fs
.sync(db_nsr
["nsd-id"])
2321 # nsr_name = db_nsr["name"] # TODO short-name??
2323 # read from db: vnf's of this ns
2324 stage
[1] = "Getting vnfrs from db."
2325 self
.logger
.debug(logging_text
+ stage
[1])
2326 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2328 # read from db: vnfd's for every vnf
2329 db_vnfds
= [] # every vnfd data
2331 # for each vnf in ns, read vnfd
2332 for vnfr
in db_vnfrs_list
:
2333 if vnfr
.get("kdur"):
2335 for kdur
in vnfr
["kdur"]:
2336 if kdur
.get("additionalParams"):
2337 kdur
["additionalParams"] = json
.loads(
2338 kdur
["additionalParams"]
2340 kdur_list
.append(kdur
)
2341 vnfr
["kdur"] = kdur_list
2343 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2344 vnfd_id
= vnfr
["vnfd-id"]
2345 vnfd_ref
= vnfr
["vnfd-ref"]
2346 self
.fs
.sync(vnfd_id
)
2348 # if we haven't this vnfd, read it from db
2349 if vnfd_id
not in db_vnfds
:
2351 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2354 self
.logger
.debug(logging_text
+ stage
[1])
2355 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2358 db_vnfds
.append(vnfd
)
2360 # Get or generates the _admin.deployed.VCA list
2361 vca_deployed_list
= None
2362 if db_nsr
["_admin"].get("deployed"):
2363 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2364 if vca_deployed_list
is None:
2365 vca_deployed_list
= []
2366 configuration_status_list
= []
2367 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2368 db_nsr_update
["configurationStatus"] = configuration_status_list
2369 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2370 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2371 elif isinstance(vca_deployed_list
, dict):
2372 # maintain backward compatibility. Change a dict to list at database
2373 vca_deployed_list
= list(vca_deployed_list
.values())
2374 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2375 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2378 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2380 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2381 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2383 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2384 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2385 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2387 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2390 # n2vc_redesign STEP 2 Deploy Network Scenario
2391 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2392 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2394 stage
[1] = "Deploying KDUs."
2395 # self.logger.debug(logging_text + "Before deploy_kdus")
2396 # Call to deploy_kdus in case exists the "vdu:kdu" param
2397 await self
.deploy_kdus(
2398 logging_text
=logging_text
,
2400 nslcmop_id
=nslcmop_id
,
2403 task_instantiation_info
=tasks_dict_info
,
2406 stage
[1] = "Getting VCA public key."
2407 # n2vc_redesign STEP 1 Get VCA public ssh-key
2408 # feature 1429. Add n2vc public key to needed VMs
2409 n2vc_key
= self
.n2vc
.get_public_key()
2410 n2vc_key_list
= [n2vc_key
]
2411 if self
.vca_config
.get("public_key"):
2412 n2vc_key_list
.append(self
.vca_config
["public_key"])
2414 stage
[1] = "Deploying NS at VIM."
2415 task_ro
= asyncio
.ensure_future(
2416 self
.instantiate_RO(
2417 logging_text
=logging_text
,
2421 db_nslcmop
=db_nslcmop
,
2424 n2vc_key_list
=n2vc_key_list
,
2428 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2429 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2431 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2432 stage
[1] = "Deploying Execution Environments."
2433 self
.logger
.debug(logging_text
+ stage
[1])
2435 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2436 for vnf_profile
in get_vnf_profiles(nsd
):
2437 vnfd_id
= vnf_profile
["vnfd-id"]
2438 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2439 member_vnf_index
= str(vnf_profile
["id"])
2440 db_vnfr
= db_vnfrs
[member_vnf_index
]
2441 base_folder
= vnfd
["_admin"]["storage"]
2447 # Get additional parameters
2448 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2449 if db_vnfr
.get("additionalParamsForVnf"):
2450 deploy_params
.update(
2451 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2454 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2455 if descriptor_config
:
2457 logging_text
=logging_text
2458 + "member_vnf_index={} ".format(member_vnf_index
),
2461 nslcmop_id
=nslcmop_id
,
2467 member_vnf_index
=member_vnf_index
,
2468 vdu_index
=vdu_index
,
2470 deploy_params
=deploy_params
,
2471 descriptor_config
=descriptor_config
,
2472 base_folder
=base_folder
,
2473 task_instantiation_info
=tasks_dict_info
,
2477 # Deploy charms for each VDU that supports one.
2478 for vdud
in get_vdu_list(vnfd
):
2480 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2481 vdur
= find_in_list(
2482 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2485 if vdur
.get("additionalParams"):
2486 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2488 deploy_params_vdu
= deploy_params
2489 deploy_params_vdu
["OSM"] = get_osm_params(
2490 db_vnfr
, vdu_id
, vdu_count_index
=0
2492 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2494 self
.logger
.debug("VDUD > {}".format(vdud
))
2496 "Descriptor config > {}".format(descriptor_config
)
2498 if descriptor_config
:
2501 for vdu_index
in range(vdud_count
):
2502 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2504 logging_text
=logging_text
2505 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2506 member_vnf_index
, vdu_id
, vdu_index
2510 nslcmop_id
=nslcmop_id
,
2516 member_vnf_index
=member_vnf_index
,
2517 vdu_index
=vdu_index
,
2519 deploy_params
=deploy_params_vdu
,
2520 descriptor_config
=descriptor_config
,
2521 base_folder
=base_folder
,
2522 task_instantiation_info
=tasks_dict_info
,
2525 for kdud
in get_kdu_list(vnfd
):
2526 kdu_name
= kdud
["name"]
2527 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2528 if descriptor_config
:
2533 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2535 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2536 if kdur
.get("additionalParams"):
2537 deploy_params_kdu
.update(
2538 parse_yaml_strings(kdur
["additionalParams"].copy())
2542 logging_text
=logging_text
,
2545 nslcmop_id
=nslcmop_id
,
2551 member_vnf_index
=member_vnf_index
,
2552 vdu_index
=vdu_index
,
2554 deploy_params
=deploy_params_kdu
,
2555 descriptor_config
=descriptor_config
,
2556 base_folder
=base_folder
,
2557 task_instantiation_info
=tasks_dict_info
,
2561 # Check if this NS has a charm configuration
2562 descriptor_config
= nsd
.get("ns-configuration")
2563 if descriptor_config
and descriptor_config
.get("juju"):
2566 member_vnf_index
= None
2572 # Get additional parameters
2573 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2574 if db_nsr
.get("additionalParamsForNs"):
2575 deploy_params
.update(
2576 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2578 base_folder
= nsd
["_admin"]["storage"]
2580 logging_text
=logging_text
,
2583 nslcmop_id
=nslcmop_id
,
2589 member_vnf_index
=member_vnf_index
,
2590 vdu_index
=vdu_index
,
2592 deploy_params
=deploy_params
,
2593 descriptor_config
=descriptor_config
,
2594 base_folder
=base_folder
,
2595 task_instantiation_info
=tasks_dict_info
,
2599 # rest of staff will be done at finally
2602 ROclient
.ROClientException
,
2608 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2611 except asyncio
.CancelledError
:
2613 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2615 exc
= "Operation was cancelled"
2616 except Exception as e
:
2617 exc
= traceback
.format_exc()
2618 self
.logger
.critical(
2619 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2624 error_list
.append(str(exc
))
2626 # wait for pending tasks
2628 stage
[1] = "Waiting for instantiate pending tasks."
2629 self
.logger
.debug(logging_text
+ stage
[1])
2630 error_list
+= await self
._wait
_for
_tasks
(
2638 stage
[1] = stage
[2] = ""
2639 except asyncio
.CancelledError
:
2640 error_list
.append("Cancelled")
2641 # TODO cancel all tasks
2642 except Exception as exc
:
2643 error_list
.append(str(exc
))
2645 # update operation-status
2646 db_nsr_update
["operational-status"] = "running"
2647 # let's begin with VCA 'configured' status (later we can change it)
2648 db_nsr_update
["config-status"] = "configured"
2649 for task
, task_name
in tasks_dict_info
.items():
2650 if not task
.done() or task
.cancelled() or task
.exception():
2651 if task_name
.startswith(self
.task_name_deploy_vca
):
2652 # A N2VC task is pending
2653 db_nsr_update
["config-status"] = "failed"
2655 # RO or KDU task is pending
2656 db_nsr_update
["operational-status"] = "failed"
2658 # update status at database
2660 error_detail
= ". ".join(error_list
)
2661 self
.logger
.error(logging_text
+ error_detail
)
2662 error_description_nslcmop
= "{} Detail: {}".format(
2663 stage
[0], error_detail
2665 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2666 nslcmop_id
, stage
[0]
2669 db_nsr_update
["detailed-status"] = (
2670 error_description_nsr
+ " Detail: " + error_detail
2672 db_nslcmop_update
["detailed-status"] = error_detail
2673 nslcmop_operation_state
= "FAILED"
2677 error_description_nsr
= error_description_nslcmop
= None
2679 db_nsr_update
["detailed-status"] = "Done"
2680 db_nslcmop_update
["detailed-status"] = "Done"
2681 nslcmop_operation_state
= "COMPLETED"
2684 self
._write
_ns
_status
(
2687 current_operation
="IDLE",
2688 current_operation_id
=None,
2689 error_description
=error_description_nsr
,
2690 error_detail
=error_detail
,
2691 other_update
=db_nsr_update
,
2693 self
._write
_op
_status
(
2696 error_message
=error_description_nslcmop
,
2697 operation_state
=nslcmop_operation_state
,
2698 other_update
=db_nslcmop_update
,
2701 if nslcmop_operation_state
:
2703 await self
.msg
.aiowrite(
2708 "nslcmop_id": nslcmop_id
,
2709 "operationState": nslcmop_operation_state
,
2713 except Exception as e
:
2715 logging_text
+ "kafka_write notification Exception {}".format(e
)
2718 self
.logger
.debug(logging_text
+ "Exit")
2719 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2721 async def _add_vca_relations(
2726 timeout
: int = 3600,
2727 vca_type
: str = None,
2732 # 1. find all relations for this VCA
2733 # 2. wait for other peers related
2737 vca_type
= vca_type
or "lxc_proxy_charm"
2739 # STEP 1: find all relations for this VCA
2742 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2743 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2746 my_vca
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))[vca_index
]
2748 # read all ns-configuration relations
2749 ns_relations
= list()
2750 db_ns_relations
= deep_get(nsd
, ("ns-configuration", "relation"))
2752 for r
in db_ns_relations
:
2753 # check if this VCA is in the relation
2754 if my_vca
.get("member-vnf-index") in (
2755 r
.get("entities")[0].get("id"),
2756 r
.get("entities")[1].get("id"),
2758 ns_relations
.append(r
)
2760 # read all vnf-configuration relations
2761 vnf_relations
= list()
2762 db_vnfd_list
= db_nsr
.get("vnfd-id")
2764 for vnfd
in db_vnfd_list
:
2765 db_vnf_relations
= None
2766 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2767 db_vnf_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
2768 if db_vnf_configuration
:
2769 db_vnf_relations
= db_vnf_configuration
.get("relation", [])
2770 if db_vnf_relations
:
2771 for r
in db_vnf_relations
:
2772 # check if this VCA is in the relation
2773 if my_vca
.get("vdu_id") in (
2774 r
.get("entities")[0].get("id"),
2775 r
.get("entities")[1].get("id"),
2777 vnf_relations
.append(r
)
2779 # if no relations, terminate
2780 if not ns_relations
and not vnf_relations
:
2781 self
.logger
.debug(logging_text
+ " No relations")
2786 + " adding relations\n {}\n {}".format(
2787 ns_relations
, vnf_relations
2796 if now
- start
>= timeout
:
2797 self
.logger
.error(logging_text
+ " : timeout adding relations")
2800 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2801 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2803 # for each defined NS relation, find the VCA's related
2804 for r
in ns_relations
.copy():
2805 from_vca_ee_id
= None
2807 from_vca_endpoint
= None
2808 to_vca_endpoint
= None
2809 vca_list
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))
2810 for vca
in vca_list
:
2811 if vca
.get("member-vnf-index") == r
.get("entities")[0].get(
2813 ) and vca
.get("config_sw_installed"):
2814 from_vca_ee_id
= vca
.get("ee_id")
2815 from_vca_endpoint
= r
.get("entities")[0].get("endpoint")
2816 if vca
.get("member-vnf-index") == r
.get("entities")[1].get(
2818 ) and vca
.get("config_sw_installed"):
2819 to_vca_ee_id
= vca
.get("ee_id")
2820 to_vca_endpoint
= r
.get("entities")[1].get("endpoint")
2821 if from_vca_ee_id
and to_vca_ee_id
:
2823 await self
.vca_map
[vca_type
].add_relation(
2824 ee_id_1
=from_vca_ee_id
,
2825 ee_id_2
=to_vca_ee_id
,
2826 endpoint_1
=from_vca_endpoint
,
2827 endpoint_2
=to_vca_endpoint
,
2830 # remove entry from relations list
2831 ns_relations
.remove(r
)
2833 # check failed peers
2835 vca_status_list
= db_nsr
.get("configurationStatus")
2837 for i
in range(len(vca_list
)):
2839 vca_status
= vca_status_list
[i
]
2840 if vca
.get("member-vnf-index") == r
.get("entities")[
2843 if vca_status
.get("status") == "BROKEN":
2844 # peer broken: remove relation from list
2845 ns_relations
.remove(r
)
2846 if vca
.get("member-vnf-index") == r
.get("entities")[
2849 if vca_status
.get("status") == "BROKEN":
2850 # peer broken: remove relation from list
2851 ns_relations
.remove(r
)
2856 # for each defined VNF relation, find the VCA's related
2857 for r
in vnf_relations
.copy():
2858 from_vca_ee_id
= None
2860 from_vca_endpoint
= None
2861 to_vca_endpoint
= None
2862 vca_list
= deep_get(db_nsr
, ("_admin", "deployed", "VCA"))
2863 for vca
in vca_list
:
2864 key_to_check
= "vdu_id"
2865 if vca
.get("vdu_id") is None:
2866 key_to_check
= "vnfd_id"
2867 if vca
.get(key_to_check
) == r
.get("entities")[0].get(
2869 ) and vca
.get("config_sw_installed"):
2870 from_vca_ee_id
= vca
.get("ee_id")
2871 from_vca_endpoint
= r
.get("entities")[0].get("endpoint")
2872 if vca
.get(key_to_check
) == r
.get("entities")[1].get(
2874 ) and vca
.get("config_sw_installed"):
2875 to_vca_ee_id
= vca
.get("ee_id")
2876 to_vca_endpoint
= r
.get("entities")[1].get("endpoint")
2877 if from_vca_ee_id
and to_vca_ee_id
:
2879 await self
.vca_map
[vca_type
].add_relation(
2880 ee_id_1
=from_vca_ee_id
,
2881 ee_id_2
=to_vca_ee_id
,
2882 endpoint_1
=from_vca_endpoint
,
2883 endpoint_2
=to_vca_endpoint
,
2886 # remove entry from relations list
2887 vnf_relations
.remove(r
)
2889 # check failed peers
2891 vca_status_list
= db_nsr
.get("configurationStatus")
2893 for i
in range(len(vca_list
)):
2895 vca_status
= vca_status_list
[i
]
2896 if vca
.get("vdu_id") == r
.get("entities")[0].get(
2899 if vca_status
.get("status") == "BROKEN":
2900 # peer broken: remove relation from list
2901 vnf_relations
.remove(r
)
2902 if vca
.get("vdu_id") == r
.get("entities")[1].get(
2905 if vca_status
.get("status") == "BROKEN":
2906 # peer broken: remove relation from list
2907 vnf_relations
.remove(r
)
2913 await asyncio
.sleep(5.0)
2915 if not ns_relations
and not vnf_relations
:
2916 self
.logger
.debug("Relations added")
2921 except Exception as e
:
2922 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
2925 async def _install_kdu(
2933 k8s_instance_info
: dict,
2934 k8params
: dict = None,
2940 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
2943 "collection": "nsrs",
2944 "filter": {"_id": nsr_id
},
2945 "path": nsr_db_path
,
2948 if k8s_instance_info
.get("kdu-deployment-name"):
2949 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
2951 kdu_instance
= self
.k8scluster_map
[
2953 ].generate_kdu_instance_name(
2954 db_dict
=db_dict_install
,
2955 kdu_model
=k8s_instance_info
["kdu-model"],
2956 kdu_name
=k8s_instance_info
["kdu-name"],
2959 # Update the nsrs table with the kdu-instance value
2963 _desc
={nsr_db_path
+ ".kdu-instance": kdu_instance
},
2966 # Update the nsrs table with the actual namespace being used, if the k8scluster-type is `juju` or
2967 # `juju-bundle`. This verification is needed because there is not a standard/homogeneous namespace
2968 # between the Helm Charts and Juju Bundles-based KNFs. If we found a way of having an homogeneous
2969 # namespace, this first verification could be removed, and the next step would be done for any kind
2971 # TODO -> find a way to have an homogeneous namespace between the Helm Charts and Juju Bundles-based
2972 # KNFs (Bug 2027: https://osm.etsi.org/bugzilla/show_bug.cgi?id=2027)
2973 if k8sclustertype
in ("juju", "juju-bundle"):
2974 # First, verify if the current namespace is present in the `_admin.projects_read` (if not, it means
2975 # that the user passed a namespace which he wants its KDU to be deployed in)
2981 "_admin.projects_write": k8s_instance_info
["namespace"],
2982 "_admin.projects_read": k8s_instance_info
["namespace"],
2988 f
"Updating namespace/model for Juju Bundle from {k8s_instance_info['namespace']} to {kdu_instance}"
2993 _desc
={f
"{nsr_db_path}.namespace": kdu_instance
},
2995 k8s_instance_info
["namespace"] = kdu_instance
2997 await self
.k8scluster_map
[k8sclustertype
].install(
2998 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2999 kdu_model
=k8s_instance_info
["kdu-model"],
3002 db_dict
=db_dict_install
,
3004 kdu_name
=k8s_instance_info
["kdu-name"],
3005 namespace
=k8s_instance_info
["namespace"],
3006 kdu_instance
=kdu_instance
,
3010 # Obtain services to obtain management service ip
3011 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
3012 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3013 kdu_instance
=kdu_instance
,
3014 namespace
=k8s_instance_info
["namespace"],
3017 # Obtain management service info (if exists)
3018 vnfr_update_dict
= {}
3019 kdu_config
= get_configuration(vnfd
, kdud
["name"])
3021 target_ee_list
= kdu_config
.get("execution-environment-list", [])
3026 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
3029 for service
in kdud
.get("service", [])
3030 if service
.get("mgmt-service")
3032 for mgmt_service
in mgmt_services
:
3033 for service
in services
:
3034 if service
["name"].startswith(mgmt_service
["name"]):
3035 # Mgmt service found, Obtain service ip
3036 ip
= service
.get("external_ip", service
.get("cluster_ip"))
3037 if isinstance(ip
, list) and len(ip
) == 1:
3041 "kdur.{}.ip-address".format(kdu_index
)
3044 # Check if must update also mgmt ip at the vnf
3045 service_external_cp
= mgmt_service
.get(
3046 "external-connection-point-ref"
3048 if service_external_cp
:
3050 deep_get(vnfd
, ("mgmt-interface", "cp"))
3051 == service_external_cp
3053 vnfr_update_dict
["ip-address"] = ip
3058 "external-connection-point-ref", ""
3060 == service_external_cp
,
3063 "kdur.{}.ip-address".format(kdu_index
)
3068 "Mgmt service name: {} not found".format(
3069 mgmt_service
["name"]
3073 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3074 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3076 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3079 and kdu_config
.get("initial-config-primitive")
3080 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3082 initial_config_primitive_list
= kdu_config
.get(
3083 "initial-config-primitive"
3085 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3087 for initial_config_primitive
in initial_config_primitive_list
:
3088 primitive_params_
= self
._map
_primitive
_params
(
3089 initial_config_primitive
, {}, {}
3092 await asyncio
.wait_for(
3093 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3094 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3095 kdu_instance
=kdu_instance
,
3096 primitive_name
=initial_config_primitive
["name"],
3097 params
=primitive_params_
,
3098 db_dict
=db_dict_install
,
3104 except Exception as e
:
3105 # Prepare update db with error and raise exception
3108 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3112 vnfr_data
.get("_id"),
3113 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3116 # ignore to keep original exception
3118 # reraise original error
3123 async def deploy_kdus(
3130 task_instantiation_info
,
3132 # Launch kdus if present in the descriptor
3134 k8scluster_id_2_uuic
= {
3135 "helm-chart-v3": {},
3140 async def _get_cluster_id(cluster_id
, cluster_type
):
3141 nonlocal k8scluster_id_2_uuic
3142 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3143 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3145 # check if K8scluster is creating and wait look if previous tasks in process
3146 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3147 "k8scluster", cluster_id
3150 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3151 task_name
, cluster_id
3153 self
.logger
.debug(logging_text
+ text
)
3154 await asyncio
.wait(task_dependency
, timeout
=3600)
3156 db_k8scluster
= self
.db
.get_one(
3157 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3159 if not db_k8scluster
:
3160 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3162 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3164 if cluster_type
== "helm-chart-v3":
3166 # backward compatibility for existing clusters that have not been initialized for helm v3
3167 k8s_credentials
= yaml
.safe_dump(
3168 db_k8scluster
.get("credentials")
3170 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3171 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3173 db_k8scluster_update
= {}
3174 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3175 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3176 db_k8scluster_update
[
3177 "_admin.helm-chart-v3.created"
3179 db_k8scluster_update
[
3180 "_admin.helm-chart-v3.operationalState"
3183 "k8sclusters", cluster_id
, db_k8scluster_update
3185 except Exception as e
:
3188 + "error initializing helm-v3 cluster: {}".format(str(e
))
3191 "K8s cluster '{}' has not been initialized for '{}'".format(
3192 cluster_id
, cluster_type
3197 "K8s cluster '{}' has not been initialized for '{}'".format(
3198 cluster_id
, cluster_type
3201 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3204 logging_text
+= "Deploy kdus: "
3207 db_nsr_update
= {"_admin.deployed.K8s": []}
3208 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3211 updated_cluster_list
= []
3212 updated_v3_cluster_list
= []
3214 for vnfr_data
in db_vnfrs
.values():
3215 vca_id
= self
.get_vca_id(vnfr_data
, {})
3216 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3217 # Step 0: Prepare and set parameters
3218 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3219 vnfd_id
= vnfr_data
.get("vnfd-id")
3220 vnfd_with_id
= find_in_list(
3221 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3225 for kdud
in vnfd_with_id
["kdu"]
3226 if kdud
["name"] == kdur
["kdu-name"]
3228 namespace
= kdur
.get("k8s-namespace")
3229 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3230 if kdur
.get("helm-chart"):
3231 kdumodel
= kdur
["helm-chart"]
3232 # Default version: helm3, if helm-version is v2 assign v2
3233 k8sclustertype
= "helm-chart-v3"
3234 self
.logger
.debug("kdur: {}".format(kdur
))
3236 kdur
.get("helm-version")
3237 and kdur
.get("helm-version") == "v2"
3239 k8sclustertype
= "helm-chart"
3240 elif kdur
.get("juju-bundle"):
3241 kdumodel
= kdur
["juju-bundle"]
3242 k8sclustertype
= "juju-bundle"
3245 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3246 "juju-bundle. Maybe an old NBI version is running".format(
3247 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3250 # check if kdumodel is a file and exists
3252 vnfd_with_id
= find_in_list(
3253 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3255 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3256 if storage
and storage
.get(
3258 ): # may be not present if vnfd has not artifacts
3259 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3260 filename
= "{}/{}/{}s/{}".format(
3266 if self
.fs
.file_exists(
3267 filename
, mode
="file"
3268 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3269 kdumodel
= self
.fs
.path
+ filename
3270 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3272 except Exception: # it is not a file
3275 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3276 step
= "Synchronize repos for k8s cluster '{}'".format(
3279 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3283 k8sclustertype
== "helm-chart"
3284 and cluster_uuid
not in updated_cluster_list
3286 k8sclustertype
== "helm-chart-v3"
3287 and cluster_uuid
not in updated_v3_cluster_list
3289 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3290 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3291 cluster_uuid
=cluster_uuid
3294 if del_repo_list
or added_repo_dict
:
3295 if k8sclustertype
== "helm-chart":
3297 "_admin.helm_charts_added." + item
: None
3298 for item
in del_repo_list
3301 "_admin.helm_charts_added." + item
: name
3302 for item
, name
in added_repo_dict
.items()
3304 updated_cluster_list
.append(cluster_uuid
)
3305 elif k8sclustertype
== "helm-chart-v3":
3307 "_admin.helm_charts_v3_added." + item
: None
3308 for item
in del_repo_list
3311 "_admin.helm_charts_v3_added." + item
: name
3312 for item
, name
in added_repo_dict
.items()
3314 updated_v3_cluster_list
.append(cluster_uuid
)
3316 logging_text
+ "repos synchronized on k8s cluster "
3317 "'{}' to_delete: {}, to_add: {}".format(
3318 k8s_cluster_id
, del_repo_list
, added_repo_dict
3323 {"_id": k8s_cluster_id
},
3329 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3330 vnfr_data
["member-vnf-index-ref"],
3334 k8s_instance_info
= {
3335 "kdu-instance": None,
3336 "k8scluster-uuid": cluster_uuid
,
3337 "k8scluster-type": k8sclustertype
,
3338 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3339 "kdu-name": kdur
["kdu-name"],
3340 "kdu-model": kdumodel
,
3341 "namespace": namespace
,
3342 "kdu-deployment-name": kdu_deployment_name
,
3344 db_path
= "_admin.deployed.K8s.{}".format(index
)
3345 db_nsr_update
[db_path
] = k8s_instance_info
3346 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3347 vnfd_with_id
= find_in_list(
3348 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3350 task
= asyncio
.ensure_future(
3359 k8params
=desc_params
,
3364 self
.lcm_tasks
.register(
3368 "instantiate_KDU-{}".format(index
),
3371 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3377 except (LcmException
, asyncio
.CancelledError
):
3379 except Exception as e
:
3380 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3381 if isinstance(e
, (N2VCException
, DbException
)):
3382 self
.logger
.error(logging_text
+ msg
)
3384 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3385 raise LcmException(msg
)
3388 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3407 task_instantiation_info
,
3410 # launch instantiate_N2VC in a asyncio task and register task object
3411 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3412 # if not found, create one entry and update database
3413 # fill db_nsr._admin.deployed.VCA.<index>
3416 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3418 if "execution-environment-list" in descriptor_config
:
3419 ee_list
= descriptor_config
.get("execution-environment-list", [])
3420 elif "juju" in descriptor_config
:
3421 ee_list
= [descriptor_config
] # ns charms
3422 else: # other types as script are not supported
3425 for ee_item
in ee_list
:
3428 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3429 ee_item
.get("juju"), ee_item
.get("helm-chart")
3432 ee_descriptor_id
= ee_item
.get("id")
3433 if ee_item
.get("juju"):
3434 vca_name
= ee_item
["juju"].get("charm")
3437 if ee_item
["juju"].get("charm") is not None
3440 if ee_item
["juju"].get("cloud") == "k8s":
3441 vca_type
= "k8s_proxy_charm"
3442 elif ee_item
["juju"].get("proxy") is False:
3443 vca_type
= "native_charm"
3444 elif ee_item
.get("helm-chart"):
3445 vca_name
= ee_item
["helm-chart"]
3446 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3449 vca_type
= "helm-v3"
3452 logging_text
+ "skipping non juju neither charm configuration"
3457 for vca_index
, vca_deployed
in enumerate(
3458 db_nsr
["_admin"]["deployed"]["VCA"]
3460 if not vca_deployed
:
3463 vca_deployed
.get("member-vnf-index") == member_vnf_index
3464 and vca_deployed
.get("vdu_id") == vdu_id
3465 and vca_deployed
.get("kdu_name") == kdu_name
3466 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3467 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3471 # not found, create one.
3473 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3476 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3478 target
+= "/kdu/{}".format(kdu_name
)
3480 "target_element": target
,
3481 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3482 "member-vnf-index": member_vnf_index
,
3484 "kdu_name": kdu_name
,
3485 "vdu_count_index": vdu_index
,
3486 "operational-status": "init", # TODO revise
3487 "detailed-status": "", # TODO revise
3488 "step": "initial-deploy", # TODO revise
3490 "vdu_name": vdu_name
,
3492 "ee_descriptor_id": ee_descriptor_id
,
3496 # create VCA and configurationStatus in db
3498 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3499 "configurationStatus.{}".format(vca_index
): dict(),
3501 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3503 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3505 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3506 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3507 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3510 task_n2vc
= asyncio
.ensure_future(
3511 self
.instantiate_N2VC(
3512 logging_text
=logging_text
,
3513 vca_index
=vca_index
,
3519 vdu_index
=vdu_index
,
3520 deploy_params
=deploy_params
,
3521 config_descriptor
=descriptor_config
,
3522 base_folder
=base_folder
,
3523 nslcmop_id
=nslcmop_id
,
3527 ee_config_descriptor
=ee_item
,
3530 self
.lcm_tasks
.register(
3534 "instantiate_N2VC-{}".format(vca_index
),
3537 task_instantiation_info
[
3539 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3540 member_vnf_index
or "", vdu_id
or ""
3544 def _create_nslcmop(nsr_id
, operation
, params
):
3546 Creates a ns-lcm-opp content to be stored at database.
3547 :param nsr_id: internal id of the instance
3548 :param operation: instantiate, terminate, scale, action, ...
3549 :param params: user parameters for the operation
3550 :return: dictionary following SOL005 format
3552 # Raise exception if invalid arguments
3553 if not (nsr_id
and operation
and params
):
3555 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3562 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3563 "operationState": "PROCESSING",
3564 "statusEnteredTime": now
,
3565 "nsInstanceId": nsr_id
,
3566 "lcmOperationType": operation
,
3568 "isAutomaticInvocation": False,
3569 "operationParams": params
,
3570 "isCancelPending": False,
3572 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3573 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3578 def _format_additional_params(self
, params
):
3579 params
= params
or {}
3580 for key
, value
in params
.items():
3581 if str(value
).startswith("!!yaml "):
3582 params
[key
] = yaml
.safe_load(value
[7:])
3585 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3586 primitive
= seq
.get("name")
3587 primitive_params
= {}
3589 "member_vnf_index": vnf_index
,
3590 "primitive": primitive
,
3591 "primitive_params": primitive_params
,
3594 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3598 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3599 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3600 if op
.get("operationState") == "COMPLETED":
3601 # b. Skip sub-operation
3602 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3603 return self
.SUBOPERATION_STATUS_SKIP
3605 # c. retry executing sub-operation
3606 # The sub-operation exists, and operationState != 'COMPLETED'
3607 # Update operationState = 'PROCESSING' to indicate a retry.
3608 operationState
= "PROCESSING"
3609 detailed_status
= "In progress"
3610 self
._update
_suboperation
_status
(
3611 db_nslcmop
, op_index
, operationState
, detailed_status
3613 # Return the sub-operation index
3614 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3615 # with arguments extracted from the sub-operation
3618 # Find a sub-operation where all keys in a matching dictionary must match
3619 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3620 def _find_suboperation(self
, db_nslcmop
, match
):
3621 if db_nslcmop
and match
:
3622 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3623 for i
, op
in enumerate(op_list
):
3624 if all(op
.get(k
) == match
[k
] for k
in match
):
3626 return self
.SUBOPERATION_STATUS_NOT_FOUND
3628 # Update status for a sub-operation given its index
3629 def _update_suboperation_status(
3630 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3632 # Update DB for HA tasks
3633 q_filter
= {"_id": db_nslcmop
["_id"]}
3635 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3636 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3639 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3642 # Add sub-operation, return the index of the added sub-operation
3643 # Optionally, set operationState, detailed-status, and operationType
3644 # Status and type are currently set for 'scale' sub-operations:
3645 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3646 # 'detailed-status' : status message
3647 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3648 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3649 def _add_suboperation(
3657 mapped_primitive_params
,
3658 operationState
=None,
3659 detailed_status
=None,
3662 RO_scaling_info
=None,
3665 return self
.SUBOPERATION_STATUS_NOT_FOUND
3666 # Get the "_admin.operations" list, if it exists
3667 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3668 op_list
= db_nslcmop_admin
.get("operations")
3669 # Create or append to the "_admin.operations" list
3671 "member_vnf_index": vnf_index
,
3673 "vdu_count_index": vdu_count_index
,
3674 "primitive": primitive
,
3675 "primitive_params": mapped_primitive_params
,
3678 new_op
["operationState"] = operationState
3680 new_op
["detailed-status"] = detailed_status
3682 new_op
["lcmOperationType"] = operationType
3684 new_op
["RO_nsr_id"] = RO_nsr_id
3686 new_op
["RO_scaling_info"] = RO_scaling_info
3688 # No existing operations, create key 'operations' with current operation as first list element
3689 db_nslcmop_admin
.update({"operations": [new_op
]})
3690 op_list
= db_nslcmop_admin
.get("operations")
3692 # Existing operations, append operation to list
3693 op_list
.append(new_op
)
3695 db_nslcmop_update
= {"_admin.operations": op_list
}
3696 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3697 op_index
= len(op_list
) - 1
3700 # Helper methods for scale() sub-operations
3702 # pre-scale/post-scale:
3703 # Check for 3 different cases:
3704 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3705 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3706 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3707 def _check_or_add_scale_suboperation(
3711 vnf_config_primitive
,
3715 RO_scaling_info
=None,
3717 # Find this sub-operation
3718 if RO_nsr_id
and RO_scaling_info
:
3719 operationType
= "SCALE-RO"
3721 "member_vnf_index": vnf_index
,
3722 "RO_nsr_id": RO_nsr_id
,
3723 "RO_scaling_info": RO_scaling_info
,
3727 "member_vnf_index": vnf_index
,
3728 "primitive": vnf_config_primitive
,
3729 "primitive_params": primitive_params
,
3730 "lcmOperationType": operationType
,
3732 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3733 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3734 # a. New sub-operation
3735 # The sub-operation does not exist, add it.
3736 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3737 # The following parameters are set to None for all kind of scaling:
3739 vdu_count_index
= None
3741 if RO_nsr_id
and RO_scaling_info
:
3742 vnf_config_primitive
= None
3743 primitive_params
= None
3746 RO_scaling_info
= None
3747 # Initial status for sub-operation
3748 operationState
= "PROCESSING"
3749 detailed_status
= "In progress"
3750 # Add sub-operation for pre/post-scaling (zero or more operations)
3751 self
._add
_suboperation
(
3757 vnf_config_primitive
,
3765 return self
.SUBOPERATION_STATUS_NEW
3767 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3768 # or op_index (operationState != 'COMPLETED')
3769 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3771 # Function to return execution_environment id
3773 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3774 # TODO vdu_index_count
3775 for vca
in vca_deployed_list
:
3776 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3779 async def destroy_N2VC(
3787 exec_primitives
=True,
3792 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3793 :param logging_text:
3795 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3796 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3797 :param vca_index: index in the database _admin.deployed.VCA
3798 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3799 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3800 not executed properly
3801 :param scaling_in: True destroys the application, False destroys the model
3802 :return: None or exception
3807 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3808 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3812 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3814 # execute terminate_primitives
3816 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3817 config_descriptor
.get("terminate-config-primitive"),
3818 vca_deployed
.get("ee_descriptor_id"),
3820 vdu_id
= vca_deployed
.get("vdu_id")
3821 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3822 vdu_name
= vca_deployed
.get("vdu_name")
3823 vnf_index
= vca_deployed
.get("member-vnf-index")
3824 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3825 for seq
in terminate_primitives
:
3826 # For each sequence in list, get primitive and call _ns_execute_primitive()
3827 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3828 vnf_index
, seq
.get("name")
3830 self
.logger
.debug(logging_text
+ step
)
3831 # Create the primitive for each sequence, i.e. "primitive": "touch"
3832 primitive
= seq
.get("name")
3833 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3838 self
._add
_suboperation
(
3845 mapped_primitive_params
,
3847 # Sub-operations: Call _ns_execute_primitive() instead of action()
3849 result
, result_detail
= await self
._ns
_execute
_primitive
(
3850 vca_deployed
["ee_id"],
3852 mapped_primitive_params
,
3856 except LcmException
:
3857 # this happens when VCA is not deployed. In this case it is not needed to terminate
3859 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3860 if result
not in result_ok
:
3862 "terminate_primitive {} for vnf_member_index={} fails with "
3863 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3865 # set that this VCA do not need terminated
3866 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3870 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3873 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3874 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3877 await self
.vca_map
[vca_type
].delete_execution_environment(
3878 vca_deployed
["ee_id"],
3879 scaling_in
=scaling_in
,
3884 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3885 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3886 namespace
= "." + db_nsr
["_id"]
3888 await self
.n2vc
.delete_namespace(
3889 namespace
=namespace
,
3890 total_timeout
=self
.timeout_charm_delete
,
3893 except N2VCNotFound
: # already deleted. Skip
3895 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3897 async def _terminate_RO(
3898 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
3901 Terminates a deployment from RO
3902 :param logging_text:
3903 :param nsr_deployed: db_nsr._admin.deployed
3906 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3907 this method will update only the index 2, but it will write on database the concatenated content of the list
3912 ro_nsr_id
= ro_delete_action
= None
3913 if nsr_deployed
and nsr_deployed
.get("RO"):
3914 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3915 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3918 stage
[2] = "Deleting ns from VIM."
3919 db_nsr_update
["detailed-status"] = " ".join(stage
)
3920 self
._write
_op
_status
(nslcmop_id
, stage
)
3921 self
.logger
.debug(logging_text
+ stage
[2])
3922 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3923 self
._write
_op
_status
(nslcmop_id
, stage
)
3924 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3925 ro_delete_action
= desc
["action_id"]
3927 "_admin.deployed.RO.nsr_delete_action_id"
3928 ] = ro_delete_action
3929 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3930 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3931 if ro_delete_action
:
3932 # wait until NS is deleted from VIM
3933 stage
[2] = "Waiting ns deleted from VIM."
3934 detailed_status_old
= None
3938 + " RO_id={} ro_delete_action={}".format(
3939 ro_nsr_id
, ro_delete_action
3942 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3943 self
._write
_op
_status
(nslcmop_id
, stage
)
3945 delete_timeout
= 20 * 60 # 20 minutes
3946 while delete_timeout
> 0:
3947 desc
= await self
.RO
.show(
3949 item_id_name
=ro_nsr_id
,
3950 extra_item
="action",
3951 extra_item_id
=ro_delete_action
,
3955 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3957 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3958 if ns_status
== "ERROR":
3959 raise ROclient
.ROClientException(ns_status_info
)
3960 elif ns_status
== "BUILD":
3961 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3962 elif ns_status
== "ACTIVE":
3963 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3964 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3969 ), "ROclient.check_action_status returns unknown {}".format(
3972 if stage
[2] != detailed_status_old
:
3973 detailed_status_old
= stage
[2]
3974 db_nsr_update
["detailed-status"] = " ".join(stage
)
3975 self
._write
_op
_status
(nslcmop_id
, stage
)
3976 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3977 await asyncio
.sleep(5, loop
=self
.loop
)
3979 else: # delete_timeout <= 0:
3980 raise ROclient
.ROClientException(
3981 "Timeout waiting ns deleted from VIM"
3984 except Exception as e
:
3985 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3987 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
3989 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3990 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3991 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3993 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
3996 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
3998 failed_detail
.append("delete conflict: {}".format(e
))
4001 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
4004 failed_detail
.append("delete error: {}".format(e
))
4006 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
4010 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
4011 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
4013 stage
[2] = "Deleting nsd from RO."
4014 db_nsr_update
["detailed-status"] = " ".join(stage
)
4015 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4016 self
._write
_op
_status
(nslcmop_id
, stage
)
4017 await self
.RO
.delete("nsd", ro_nsd_id
)
4019 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
4021 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4022 except Exception as e
:
4024 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4026 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4028 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
4031 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4033 failed_detail
.append(
4034 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
4036 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4038 failed_detail
.append(
4039 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
4041 self
.logger
.error(logging_text
+ failed_detail
[-1])
4043 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
4044 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
4045 if not vnf_deployed
or not vnf_deployed
["id"]:
4048 ro_vnfd_id
= vnf_deployed
["id"]
4051 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
4052 vnf_deployed
["member-vnf-index"], ro_vnfd_id
4054 db_nsr_update
["detailed-status"] = " ".join(stage
)
4055 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4056 self
._write
_op
_status
(nslcmop_id
, stage
)
4057 await self
.RO
.delete("vnfd", ro_vnfd_id
)
4059 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
4061 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
4062 except Exception as e
:
4064 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4067 "_admin.deployed.RO.vnfd.{}.id".format(index
)
4071 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4074 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4076 failed_detail
.append(
4077 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4079 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4081 failed_detail
.append(
4082 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4084 self
.logger
.error(logging_text
+ failed_detail
[-1])
4087 stage
[2] = "Error deleting from VIM"
4089 stage
[2] = "Deleted from VIM"
4090 db_nsr_update
["detailed-status"] = " ".join(stage
)
4091 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4092 self
._write
_op
_status
(nslcmop_id
, stage
)
4095 raise LcmException("; ".join(failed_detail
))
4097 async def terminate(self
, nsr_id
, nslcmop_id
):
4098 # Try to lock HA task here
4099 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4100 if not task_is_locked_by_me
:
4103 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4104 self
.logger
.debug(logging_text
+ "Enter")
4105 timeout_ns_terminate
= self
.timeout_ns_terminate
4108 operation_params
= None
4110 error_list
= [] # annotates all failed error messages
4111 db_nslcmop_update
= {}
4112 autoremove
= False # autoremove after terminated
4113 tasks_dict_info
= {}
4116 "Stage 1/3: Preparing task.",
4117 "Waiting for previous operations to terminate.",
4120 # ^ contains [stage, step, VIM-status]
4122 # wait for any previous tasks in process
4123 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4125 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4126 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4127 operation_params
= db_nslcmop
.get("operationParams") or {}
4128 if operation_params
.get("timeout_ns_terminate"):
4129 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4130 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4131 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4133 db_nsr_update
["operational-status"] = "terminating"
4134 db_nsr_update
["config-status"] = "terminating"
4135 self
._write
_ns
_status
(
4137 ns_state
="TERMINATING",
4138 current_operation
="TERMINATING",
4139 current_operation_id
=nslcmop_id
,
4140 other_update
=db_nsr_update
,
4142 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4143 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4144 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4147 stage
[1] = "Getting vnf descriptors from db."
4148 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4150 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4152 db_vnfds_from_id
= {}
4153 db_vnfds_from_member_index
= {}
4155 for vnfr
in db_vnfrs_list
:
4156 vnfd_id
= vnfr
["vnfd-id"]
4157 if vnfd_id
not in db_vnfds_from_id
:
4158 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4159 db_vnfds_from_id
[vnfd_id
] = vnfd
4160 db_vnfds_from_member_index
[
4161 vnfr
["member-vnf-index-ref"]
4162 ] = db_vnfds_from_id
[vnfd_id
]
4164 # Destroy individual execution environments when there are terminating primitives.
4165 # Rest of EE will be deleted at once
4166 # TODO - check before calling _destroy_N2VC
4167 # if not operation_params.get("skip_terminate_primitives"):#
4168 # or not vca.get("needed_terminate"):
4169 stage
[0] = "Stage 2/3 execute terminating primitives."
4170 self
.logger
.debug(logging_text
+ stage
[0])
4171 stage
[1] = "Looking execution environment that needs terminate."
4172 self
.logger
.debug(logging_text
+ stage
[1])
4174 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4175 config_descriptor
= None
4176 vca_member_vnf_index
= vca
.get("member-vnf-index")
4177 vca_id
= self
.get_vca_id(
4178 db_vnfrs_dict
.get(vca_member_vnf_index
)
4179 if vca_member_vnf_index
4183 if not vca
or not vca
.get("ee_id"):
4185 if not vca
.get("member-vnf-index"):
4187 config_descriptor
= db_nsr
.get("ns-configuration")
4188 elif vca
.get("vdu_id"):
4189 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4190 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4191 elif vca
.get("kdu_name"):
4192 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4193 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4195 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4196 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4197 vca_type
= vca
.get("type")
4198 exec_terminate_primitives
= not operation_params
.get(
4199 "skip_terminate_primitives"
4200 ) and vca
.get("needed_terminate")
4201 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4202 # pending native charms
4204 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4206 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4207 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4208 task
= asyncio
.ensure_future(
4216 exec_terminate_primitives
,
4220 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4222 # wait for pending tasks of terminate primitives
4226 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4228 error_list
= await self
._wait
_for
_tasks
(
4231 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4235 tasks_dict_info
.clear()
4237 return # raise LcmException("; ".join(error_list))
4239 # remove All execution environments at once
4240 stage
[0] = "Stage 3/3 delete all."
4242 if nsr_deployed
.get("VCA"):
4243 stage
[1] = "Deleting all execution environments."
4244 self
.logger
.debug(logging_text
+ stage
[1])
4245 vca_id
= self
.get_vca_id({}, db_nsr
)
4246 task_delete_ee
= asyncio
.ensure_future(
4248 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4249 timeout
=self
.timeout_charm_delete
,
4252 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4253 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4255 # Delete from k8scluster
4256 stage
[1] = "Deleting KDUs."
4257 self
.logger
.debug(logging_text
+ stage
[1])
4258 # print(nsr_deployed)
4259 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4260 if not kdu
or not kdu
.get("kdu-instance"):
4262 kdu_instance
= kdu
.get("kdu-instance")
4263 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4264 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4265 vca_id
= self
.get_vca_id({}, db_nsr
)
4266 task_delete_kdu_instance
= asyncio
.ensure_future(
4267 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4268 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4269 kdu_instance
=kdu_instance
,
4276 + "Unknown k8s deployment type {}".format(
4277 kdu
.get("k8scluster-type")
4282 task_delete_kdu_instance
4283 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4286 stage
[1] = "Deleting ns from VIM."
4288 task_delete_ro
= asyncio
.ensure_future(
4289 self
._terminate
_ng
_ro
(
4290 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4294 task_delete_ro
= asyncio
.ensure_future(
4296 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4299 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4301 # rest of staff will be done at finally
4304 ROclient
.ROClientException
,
4309 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4311 except asyncio
.CancelledError
:
4313 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4315 exc
= "Operation was cancelled"
4316 except Exception as e
:
4317 exc
= traceback
.format_exc()
4318 self
.logger
.critical(
4319 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4324 error_list
.append(str(exc
))
4326 # wait for pending tasks
4328 stage
[1] = "Waiting for terminate pending tasks."
4329 self
.logger
.debug(logging_text
+ stage
[1])
4330 error_list
+= await self
._wait
_for
_tasks
(
4333 timeout_ns_terminate
,
4337 stage
[1] = stage
[2] = ""
4338 except asyncio
.CancelledError
:
4339 error_list
.append("Cancelled")
4340 # TODO cancell all tasks
4341 except Exception as exc
:
4342 error_list
.append(str(exc
))
4343 # update status at database
4345 error_detail
= "; ".join(error_list
)
4346 # self.logger.error(logging_text + error_detail)
4347 error_description_nslcmop
= "{} Detail: {}".format(
4348 stage
[0], error_detail
4350 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4351 nslcmop_id
, stage
[0]
4354 db_nsr_update
["operational-status"] = "failed"
4355 db_nsr_update
["detailed-status"] = (
4356 error_description_nsr
+ " Detail: " + error_detail
4358 db_nslcmop_update
["detailed-status"] = error_detail
4359 nslcmop_operation_state
= "FAILED"
4363 error_description_nsr
= error_description_nslcmop
= None
4364 ns_state
= "NOT_INSTANTIATED"
4365 db_nsr_update
["operational-status"] = "terminated"
4366 db_nsr_update
["detailed-status"] = "Done"
4367 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4368 db_nslcmop_update
["detailed-status"] = "Done"
4369 nslcmop_operation_state
= "COMPLETED"
4372 self
._write
_ns
_status
(
4375 current_operation
="IDLE",
4376 current_operation_id
=None,
4377 error_description
=error_description_nsr
,
4378 error_detail
=error_detail
,
4379 other_update
=db_nsr_update
,
4381 self
._write
_op
_status
(
4384 error_message
=error_description_nslcmop
,
4385 operation_state
=nslcmop_operation_state
,
4386 other_update
=db_nslcmop_update
,
4388 if ns_state
== "NOT_INSTANTIATED":
4392 {"nsr-id-ref": nsr_id
},
4393 {"_admin.nsState": "NOT_INSTANTIATED"},
4395 except DbException
as e
:
4398 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4402 if operation_params
:
4403 autoremove
= operation_params
.get("autoremove", False)
4404 if nslcmop_operation_state
:
4406 await self
.msg
.aiowrite(
4411 "nslcmop_id": nslcmop_id
,
4412 "operationState": nslcmop_operation_state
,
4413 "autoremove": autoremove
,
4417 except Exception as e
:
4419 logging_text
+ "kafka_write notification Exception {}".format(e
)
4422 self
.logger
.debug(logging_text
+ "Exit")
4423 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4425 async def _wait_for_tasks(
4426 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4429 error_detail_list
= []
4431 pending_tasks
= list(created_tasks_info
.keys())
4432 num_tasks
= len(pending_tasks
)
4434 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4435 self
._write
_op
_status
(nslcmop_id
, stage
)
4436 while pending_tasks
:
4438 _timeout
= timeout
+ time_start
- time()
4439 done
, pending_tasks
= await asyncio
.wait(
4440 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4442 num_done
+= len(done
)
4443 if not done
: # Timeout
4444 for task
in pending_tasks
:
4445 new_error
= created_tasks_info
[task
] + ": Timeout"
4446 error_detail_list
.append(new_error
)
4447 error_list
.append(new_error
)
4450 if task
.cancelled():
4453 exc
= task
.exception()
4455 if isinstance(exc
, asyncio
.TimeoutError
):
4457 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4458 error_list
.append(created_tasks_info
[task
])
4459 error_detail_list
.append(new_error
)
4466 ROclient
.ROClientException
,
4472 self
.logger
.error(logging_text
+ new_error
)
4474 exc_traceback
= "".join(
4475 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4479 + created_tasks_info
[task
]
4485 logging_text
+ created_tasks_info
[task
] + ": Done"
4487 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4489 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4490 if nsr_id
: # update also nsr
4495 "errorDescription": "Error at: " + ", ".join(error_list
),
4496 "errorDetail": ". ".join(error_detail_list
),
4499 self
._write
_op
_status
(nslcmop_id
, stage
)
4500 return error_detail_list
4503 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4505 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4506 The default-value is used. If it is between < > it look for a value at instantiation_params
4507 :param primitive_desc: portion of VNFD/NSD that describes primitive
4508 :param params: Params provided by user
4509 :param instantiation_params: Instantiation params provided by user
4510 :return: a dictionary with the calculated params
4512 calculated_params
= {}
4513 for parameter
in primitive_desc
.get("parameter", ()):
4514 param_name
= parameter
["name"]
4515 if param_name
in params
:
4516 calculated_params
[param_name
] = params
[param_name
]
4517 elif "default-value" in parameter
or "value" in parameter
:
4518 if "value" in parameter
:
4519 calculated_params
[param_name
] = parameter
["value"]
4521 calculated_params
[param_name
] = parameter
["default-value"]
4523 isinstance(calculated_params
[param_name
], str)
4524 and calculated_params
[param_name
].startswith("<")
4525 and calculated_params
[param_name
].endswith(">")
4527 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4528 calculated_params
[param_name
] = instantiation_params
[
4529 calculated_params
[param_name
][1:-1]
4533 "Parameter {} needed to execute primitive {} not provided".format(
4534 calculated_params
[param_name
], primitive_desc
["name"]
4539 "Parameter {} needed to execute primitive {} not provided".format(
4540 param_name
, primitive_desc
["name"]
4544 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4545 calculated_params
[param_name
] = yaml
.safe_dump(
4546 calculated_params
[param_name
], default_flow_style
=True, width
=256
4548 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4550 ].startswith("!!yaml "):
4551 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4552 if parameter
.get("data-type") == "INTEGER":
4554 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4555 except ValueError: # error converting string to int
4557 "Parameter {} of primitive {} must be integer".format(
4558 param_name
, primitive_desc
["name"]
4561 elif parameter
.get("data-type") == "BOOLEAN":
4562 calculated_params
[param_name
] = not (
4563 (str(calculated_params
[param_name
])).lower() == "false"
4566 # add always ns_config_info if primitive name is config
4567 if primitive_desc
["name"] == "config":
4568 if "ns_config_info" in instantiation_params
:
4569 calculated_params
["ns_config_info"] = instantiation_params
[
4572 return calculated_params
4574 def _look_for_deployed_vca(
4581 ee_descriptor_id
=None,
4583 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4584 for vca
in deployed_vca
:
4587 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4590 vdu_count_index
is not None
4591 and vdu_count_index
!= vca
["vdu_count_index"]
4594 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4596 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4600 # vca_deployed not found
4602 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4603 " is not deployed".format(
4612 ee_id
= vca
.get("ee_id")
4614 "type", "lxc_proxy_charm"
4615 ) # default value for backward compatibility - proxy charm
4618 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4619 "execution environment".format(
4620 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4623 return ee_id
, vca_type
4625 async def _ns_execute_primitive(
4631 retries_interval
=30,
4638 if primitive
== "config":
4639 primitive_params
= {"params": primitive_params
}
4641 vca_type
= vca_type
or "lxc_proxy_charm"
4645 output
= await asyncio
.wait_for(
4646 self
.vca_map
[vca_type
].exec_primitive(
4648 primitive_name
=primitive
,
4649 params_dict
=primitive_params
,
4650 progress_timeout
=self
.timeout_progress_primitive
,
4651 total_timeout
=self
.timeout_primitive
,
4656 timeout
=timeout
or self
.timeout_primitive
,
4660 except asyncio
.CancelledError
:
4662 except Exception as e
: # asyncio.TimeoutError
4663 if isinstance(e
, asyncio
.TimeoutError
):
4668 "Error executing action {} on {} -> {}".format(
4673 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4675 return "FAILED", str(e
)
4677 return "COMPLETED", output
4679 except (LcmException
, asyncio
.CancelledError
):
4681 except Exception as e
:
4682 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4684 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4686 Updating the vca_status with latest juju information in nsrs record
4687 :param: nsr_id: Id of the nsr
4688 :param: nslcmop_id: Id of the nslcmop
4692 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4693 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4694 vca_id
= self
.get_vca_id({}, db_nsr
)
4695 if db_nsr
["_admin"]["deployed"]["K8s"]:
4696 for _
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4697 cluster_uuid
, kdu_instance
, cluster_type
= (
4698 k8s
["k8scluster-uuid"],
4699 k8s
["kdu-instance"],
4700 k8s
["k8scluster-type"],
4702 await self
._on
_update
_k
8s
_db
(
4703 cluster_uuid
=cluster_uuid
,
4704 kdu_instance
=kdu_instance
,
4705 filter={"_id": nsr_id
},
4707 cluster_type
=cluster_type
,
4710 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4711 table
, filter = "nsrs", {"_id": nsr_id
}
4712 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4713 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4715 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4716 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4718 async def action(self
, nsr_id
, nslcmop_id
):
4719 # Try to lock HA task here
4720 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4721 if not task_is_locked_by_me
:
4724 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4725 self
.logger
.debug(logging_text
+ "Enter")
4726 # get all needed from database
4730 db_nslcmop_update
= {}
4731 nslcmop_operation_state
= None
4732 error_description_nslcmop
= None
4735 # wait for any previous tasks in process
4736 step
= "Waiting for previous operations to terminate"
4737 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4739 self
._write
_ns
_status
(
4742 current_operation
="RUNNING ACTION",
4743 current_operation_id
=nslcmop_id
,
4746 step
= "Getting information from database"
4747 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4748 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4749 if db_nslcmop
["operationParams"].get("primitive_params"):
4750 db_nslcmop
["operationParams"]["primitive_params"] = json
.loads(
4751 db_nslcmop
["operationParams"]["primitive_params"]
4754 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4755 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4756 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4757 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4758 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4759 primitive
= db_nslcmop
["operationParams"]["primitive"]
4760 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4761 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4762 "timeout_ns_action", self
.timeout_primitive
4766 step
= "Getting vnfr from database"
4767 db_vnfr
= self
.db
.get_one(
4768 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4770 if db_vnfr
.get("kdur"):
4772 for kdur
in db_vnfr
["kdur"]:
4773 if kdur
.get("additionalParams"):
4774 kdur
["additionalParams"] = json
.loads(
4775 kdur
["additionalParams"]
4777 kdur_list
.append(kdur
)
4778 db_vnfr
["kdur"] = kdur_list
4779 step
= "Getting vnfd from database"
4780 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4782 step
= "Getting nsd from database"
4783 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4785 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4786 # for backward compatibility
4787 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4788 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4789 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4790 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4792 # look for primitive
4793 config_primitive_desc
= descriptor_configuration
= None
4795 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4797 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4799 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4801 descriptor_configuration
= db_nsd
.get("ns-configuration")
4803 if descriptor_configuration
and descriptor_configuration
.get(
4806 for config_primitive
in descriptor_configuration
["config-primitive"]:
4807 if config_primitive
["name"] == primitive
:
4808 config_primitive_desc
= config_primitive
4811 if not config_primitive_desc
:
4812 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4814 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4818 primitive_name
= primitive
4819 ee_descriptor_id
= None
4821 primitive_name
= config_primitive_desc
.get(
4822 "execution-environment-primitive", primitive
4824 ee_descriptor_id
= config_primitive_desc
.get(
4825 "execution-environment-ref"
4831 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4833 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4836 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4838 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4840 desc_params
= parse_yaml_strings(
4841 db_vnfr
.get("additionalParamsForVnf")
4844 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4845 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4846 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4848 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4849 actions
.add(primitive
["name"])
4850 for primitive
in kdu_configuration
.get("config-primitive", []):
4851 actions
.add(primitive
["name"])
4852 kdu_action
= True if primitive_name
in actions
else False
4854 # TODO check if ns is in a proper status
4856 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4858 # kdur and desc_params already set from before
4859 if primitive_params
:
4860 desc_params
.update(primitive_params
)
4861 # TODO Check if we will need something at vnf level
4862 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4864 kdu_name
== kdu
["kdu-name"]
4865 and kdu
["member-vnf-index"] == vnf_index
4870 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4873 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4874 msg
= "unknown k8scluster-type '{}'".format(
4875 kdu
.get("k8scluster-type")
4877 raise LcmException(msg
)
4880 "collection": "nsrs",
4881 "filter": {"_id": nsr_id
},
4882 "path": "_admin.deployed.K8s.{}".format(index
),
4886 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4888 step
= "Executing kdu {}".format(primitive_name
)
4889 if primitive_name
== "upgrade":
4890 if desc_params
.get("kdu_model"):
4891 kdu_model
= desc_params
.get("kdu_model")
4892 del desc_params
["kdu_model"]
4894 kdu_model
= kdu
.get("kdu-model")
4895 parts
= kdu_model
.split(sep
=":")
4897 kdu_model
= parts
[0]
4899 detailed_status
= await asyncio
.wait_for(
4900 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4901 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4902 kdu_instance
=kdu
.get("kdu-instance"),
4904 kdu_model
=kdu_model
,
4907 timeout
=timeout_ns_action
,
4909 timeout
=timeout_ns_action
+ 10,
4912 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4914 elif primitive_name
== "rollback":
4915 detailed_status
= await asyncio
.wait_for(
4916 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4917 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4918 kdu_instance
=kdu
.get("kdu-instance"),
4921 timeout
=timeout_ns_action
,
4923 elif primitive_name
== "status":
4924 detailed_status
= await asyncio
.wait_for(
4925 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4926 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4927 kdu_instance
=kdu
.get("kdu-instance"),
4930 timeout
=timeout_ns_action
,
4933 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
4934 kdu
["kdu-name"], nsr_id
4936 params
= self
._map
_primitive
_params
(
4937 config_primitive_desc
, primitive_params
, desc_params
4940 detailed_status
= await asyncio
.wait_for(
4941 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
4942 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4943 kdu_instance
=kdu_instance
,
4944 primitive_name
=primitive_name
,
4947 timeout
=timeout_ns_action
,
4950 timeout
=timeout_ns_action
,
4954 nslcmop_operation_state
= "COMPLETED"
4956 detailed_status
= ""
4957 nslcmop_operation_state
= "FAILED"
4959 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
4960 nsr_deployed
["VCA"],
4961 member_vnf_index
=vnf_index
,
4963 vdu_count_index
=vdu_count_index
,
4964 ee_descriptor_id
=ee_descriptor_id
,
4966 for vca_index
, vca_deployed
in enumerate(
4967 db_nsr
["_admin"]["deployed"]["VCA"]
4969 if vca_deployed
.get("member-vnf-index") == vnf_index
:
4971 "collection": "nsrs",
4972 "filter": {"_id": nsr_id
},
4973 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
4977 nslcmop_operation_state
,
4979 ) = await self
._ns
_execute
_primitive
(
4981 primitive
=primitive_name
,
4982 primitive_params
=self
._map
_primitive
_params
(
4983 config_primitive_desc
, primitive_params
, desc_params
4985 timeout
=timeout_ns_action
,
4991 db_nslcmop_update
["detailed-status"] = detailed_status
4992 error_description_nslcmop
= (
4993 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
4997 + " task Done with result {} {}".format(
4998 nslcmop_operation_state
, detailed_status
5001 return # database update is called inside finally
5003 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5004 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5006 except asyncio
.CancelledError
:
5008 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5010 exc
= "Operation was cancelled"
5011 except asyncio
.TimeoutError
:
5012 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5014 except Exception as e
:
5015 exc
= traceback
.format_exc()
5016 self
.logger
.critical(
5017 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5026 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5027 nslcmop_operation_state
= "FAILED"
5029 self
._write
_ns
_status
(
5033 ], # TODO check if degraded. For the moment use previous status
5034 current_operation
="IDLE",
5035 current_operation_id
=None,
5036 # error_description=error_description_nsr,
5037 # error_detail=error_detail,
5038 other_update
=db_nsr_update
,
5041 self
._write
_op
_status
(
5044 error_message
=error_description_nslcmop
,
5045 operation_state
=nslcmop_operation_state
,
5046 other_update
=db_nslcmop_update
,
5049 if nslcmop_operation_state
:
5051 await self
.msg
.aiowrite(
5056 "nslcmop_id": nslcmop_id
,
5057 "operationState": nslcmop_operation_state
,
5061 except Exception as e
:
5063 logging_text
+ "kafka_write notification Exception {}".format(e
)
5065 self
.logger
.debug(logging_text
+ "Exit")
5066 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5067 return nslcmop_operation_state
, detailed_status
5069 async def scale(self
, nsr_id
, nslcmop_id
):
5070 # Try to lock HA task here
5071 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5072 if not task_is_locked_by_me
:
5075 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5076 stage
= ["", "", ""]
5077 tasks_dict_info
= {}
5078 # ^ stage, step, VIM progress
5079 self
.logger
.debug(logging_text
+ "Enter")
5080 # get all needed from database
5082 db_nslcmop_update
= {}
5085 # in case of error, indicates what part of scale was failed to put nsr at error status
5086 scale_process
= None
5087 old_operational_status
= ""
5088 old_config_status
= ""
5091 # wait for any previous tasks in process
5092 step
= "Waiting for previous operations to terminate"
5093 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5094 self
._write
_ns
_status
(
5097 current_operation
="SCALING",
5098 current_operation_id
=nslcmop_id
,
5101 step
= "Getting nslcmop from database"
5103 step
+ " after having waited for previous tasks to be completed"
5105 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5107 step
= "Getting nsr from database"
5108 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5109 old_operational_status
= db_nsr
["operational-status"]
5110 old_config_status
= db_nsr
["config-status"]
5112 step
= "Parsing scaling parameters"
5113 db_nsr_update
["operational-status"] = "scaling"
5114 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5115 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5117 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5119 ]["member-vnf-index"]
5120 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5122 ]["scaling-group-descriptor"]
5123 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5124 # for backward compatibility
5125 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5126 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5127 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5128 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5130 step
= "Getting vnfr from database"
5131 db_vnfr
= self
.db
.get_one(
5132 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5135 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5137 step
= "Getting vnfd from database"
5138 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5140 base_folder
= db_vnfd
["_admin"]["storage"]
5142 step
= "Getting scaling-group-descriptor"
5143 scaling_descriptor
= find_in_list(
5144 get_scaling_aspect(db_vnfd
),
5145 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5147 if not scaling_descriptor
:
5149 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5150 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5153 step
= "Sending scale order to VIM"
5154 # TODO check if ns is in a proper status
5156 if not db_nsr
["_admin"].get("scaling-group"):
5161 "_admin.scaling-group": [
5162 {"name": scaling_group
, "nb-scale-op": 0}
5166 admin_scale_index
= 0
5168 for admin_scale_index
, admin_scale_info
in enumerate(
5169 db_nsr
["_admin"]["scaling-group"]
5171 if admin_scale_info
["name"] == scaling_group
:
5172 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5174 else: # not found, set index one plus last element and add new entry with the name
5175 admin_scale_index
+= 1
5177 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5180 vca_scaling_info
= []
5181 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5182 if scaling_type
== "SCALE_OUT":
5183 if "aspect-delta-details" not in scaling_descriptor
:
5185 "Aspect delta details not fount in scaling descriptor {}".format(
5186 scaling_descriptor
["name"]
5189 # count if max-instance-count is reached
5190 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5192 scaling_info
["scaling_direction"] = "OUT"
5193 scaling_info
["vdu-create"] = {}
5194 scaling_info
["kdu-create"] = {}
5195 for delta
in deltas
:
5196 for vdu_delta
in delta
.get("vdu-delta", {}):
5197 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5198 # vdu_index also provides the number of instance of the targeted vdu
5199 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5200 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5204 additional_params
= (
5205 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5208 cloud_init_list
= []
5210 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5211 max_instance_count
= 10
5212 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5213 max_instance_count
= vdu_profile
.get(
5214 "max-number-of-instances", 10
5217 default_instance_num
= get_number_of_instances(
5220 instances_number
= vdu_delta
.get("number-of-instances", 1)
5221 nb_scale_op
+= instances_number
5223 new_instance_count
= nb_scale_op
+ default_instance_num
5224 # Control if new count is over max and vdu count is less than max.
5225 # Then assign new instance count
5226 if new_instance_count
> max_instance_count
> vdu_count
:
5227 instances_number
= new_instance_count
- max_instance_count
5229 instances_number
= instances_number
5231 if new_instance_count
> max_instance_count
:
5233 "reached the limit of {} (max-instance-count) "
5234 "scaling-out operations for the "
5235 "scaling-group-descriptor '{}'".format(
5236 nb_scale_op
, scaling_group
5239 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5241 # TODO Information of its own ip is not available because db_vnfr is not updated.
5242 additional_params
["OSM"] = get_osm_params(
5243 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5245 cloud_init_list
.append(
5246 self
._parse
_cloud
_init
(
5253 vca_scaling_info
.append(
5255 "osm_vdu_id": vdu_delta
["id"],
5256 "member-vnf-index": vnf_index
,
5258 "vdu_index": vdu_index
+ x
,
5261 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5262 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5263 kdu_profile
= get_kdu_profile(db_vnfd
, kdu_delta
["id"])
5264 kdu_name
= kdu_profile
["kdu-name"]
5265 resource_name
= kdu_profile
["resource-name"]
5267 # Might have different kdus in the same delta
5268 # Should have list for each kdu
5269 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5270 scaling_info
["kdu-create"][kdu_name
] = []
5272 kdur
= get_kdur(db_vnfr
, kdu_name
)
5273 if kdur
.get("helm-chart"):
5274 k8s_cluster_type
= "helm-chart-v3"
5275 self
.logger
.debug("kdur: {}".format(kdur
))
5277 kdur
.get("helm-version")
5278 and kdur
.get("helm-version") == "v2"
5280 k8s_cluster_type
= "helm-chart"
5281 raise NotImplementedError
5282 elif kdur
.get("juju-bundle"):
5283 k8s_cluster_type
= "juju-bundle"
5286 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5287 "juju-bundle. Maybe an old NBI version is running".format(
5288 db_vnfr
["member-vnf-index-ref"], kdu_name
5292 max_instance_count
= 10
5293 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5294 max_instance_count
= kdu_profile
.get(
5295 "max-number-of-instances", 10
5298 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5299 deployed_kdu
, _
= get_deployed_kdu(
5300 nsr_deployed
, kdu_name
, vnf_index
5302 if deployed_kdu
is None:
5304 "KDU '{}' for vnf '{}' not deployed".format(
5308 kdu_instance
= deployed_kdu
.get("kdu-instance")
5309 instance_num
= await self
.k8scluster_map
[
5311 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5312 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5313 "number-of-instances", 1
5316 # Control if new count is over max and instance_num is less than max.
5317 # Then assign max instance number to kdu replica count
5318 if kdu_replica_count
> max_instance_count
> instance_num
:
5319 kdu_replica_count
= max_instance_count
5320 if kdu_replica_count
> max_instance_count
:
5322 "reached the limit of {} (max-instance-count) "
5323 "scaling-out operations for the "
5324 "scaling-group-descriptor '{}'".format(
5325 instance_num
, scaling_group
5329 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5330 vca_scaling_info
.append(
5332 "osm_kdu_id": kdu_name
,
5333 "member-vnf-index": vnf_index
,
5335 "kdu_index": instance_num
+ x
- 1,
5338 scaling_info
["kdu-create"][kdu_name
].append(
5340 "member-vnf-index": vnf_index
,
5342 "k8s-cluster-type": k8s_cluster_type
,
5343 "resource-name": resource_name
,
5344 "scale": kdu_replica_count
,
5347 elif scaling_type
== "SCALE_IN":
5348 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5350 scaling_info
["scaling_direction"] = "IN"
5351 scaling_info
["vdu-delete"] = {}
5352 scaling_info
["kdu-delete"] = {}
5354 for delta
in deltas
:
5355 for vdu_delta
in delta
.get("vdu-delta", {}):
5356 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5357 min_instance_count
= 0
5358 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5359 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5360 min_instance_count
= vdu_profile
["min-number-of-instances"]
5362 default_instance_num
= get_number_of_instances(
5363 db_vnfd
, vdu_delta
["id"]
5365 instance_num
= vdu_delta
.get("number-of-instances", 1)
5366 nb_scale_op
-= instance_num
5368 new_instance_count
= nb_scale_op
+ default_instance_num
5370 if new_instance_count
< min_instance_count
< vdu_count
:
5371 instances_number
= min_instance_count
- new_instance_count
5373 instances_number
= instance_num
5375 if new_instance_count
< min_instance_count
:
5377 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5378 "scaling-group-descriptor '{}'".format(
5379 nb_scale_op
, scaling_group
5382 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5383 vca_scaling_info
.append(
5385 "osm_vdu_id": vdu_delta
["id"],
5386 "member-vnf-index": vnf_index
,
5388 "vdu_index": vdu_index
- 1 - x
,
5391 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5392 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5393 kdu_profile
= get_kdu_profile(db_vnfd
, kdu_delta
["id"])
5394 kdu_name
= kdu_profile
["kdu-name"]
5395 resource_name
= kdu_profile
["resource-name"]
5397 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5398 scaling_info
["kdu-delete"][kdu_name
] = []
5400 kdur
= get_kdur(db_vnfr
, kdu_name
)
5401 if kdur
.get("helm-chart"):
5402 k8s_cluster_type
= "helm-chart-v3"
5403 self
.logger
.debug("kdur: {}".format(kdur
))
5405 kdur
.get("helm-version")
5406 and kdur
.get("helm-version") == "v2"
5408 k8s_cluster_type
= "helm-chart"
5409 raise NotImplementedError
5410 elif kdur
.get("juju-bundle"):
5411 k8s_cluster_type
= "juju-bundle"
5414 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5415 "juju-bundle. Maybe an old NBI version is running".format(
5416 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5420 min_instance_count
= 0
5421 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5422 min_instance_count
= kdu_profile
["min-number-of-instances"]
5424 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5425 deployed_kdu
, _
= get_deployed_kdu(
5426 nsr_deployed
, kdu_name
, vnf_index
5428 if deployed_kdu
is None:
5430 "KDU '{}' for vnf '{}' not deployed".format(
5434 kdu_instance
= deployed_kdu
.get("kdu-instance")
5435 instance_num
= await self
.k8scluster_map
[
5437 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5438 kdu_replica_count
= instance_num
- kdu_delta
.get(
5439 "number-of-instances", 1
5442 if kdu_replica_count
< min_instance_count
< instance_num
:
5443 kdu_replica_count
= min_instance_count
5444 if kdu_replica_count
< min_instance_count
:
5446 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5447 "scaling-group-descriptor '{}'".format(
5448 instance_num
, scaling_group
5452 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5453 vca_scaling_info
.append(
5455 "osm_kdu_id": kdu_name
,
5456 "member-vnf-index": vnf_index
,
5458 "kdu_index": instance_num
- x
- 1,
5461 scaling_info
["kdu-delete"][kdu_name
].append(
5463 "member-vnf-index": vnf_index
,
5465 "k8s-cluster-type": k8s_cluster_type
,
5466 "resource-name": resource_name
,
5467 "scale": kdu_replica_count
,
5471 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5472 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5473 if scaling_info
["scaling_direction"] == "IN":
5474 for vdur
in reversed(db_vnfr
["vdur"]):
5475 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5476 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5477 scaling_info
["vdu"].append(
5479 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5480 "vdu_id": vdur
["vdu-id-ref"],
5484 for interface
in vdur
["interfaces"]:
5485 scaling_info
["vdu"][-1]["interface"].append(
5487 "name": interface
["name"],
5488 "ip_address": interface
["ip-address"],
5489 "mac_address": interface
.get("mac-address"),
5492 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5495 step
= "Executing pre-scale vnf-config-primitive"
5496 if scaling_descriptor
.get("scaling-config-action"):
5497 for scaling_config_action
in scaling_descriptor
[
5498 "scaling-config-action"
5501 scaling_config_action
.get("trigger") == "pre-scale-in"
5502 and scaling_type
== "SCALE_IN"
5504 scaling_config_action
.get("trigger") == "pre-scale-out"
5505 and scaling_type
== "SCALE_OUT"
5507 vnf_config_primitive
= scaling_config_action
[
5508 "vnf-config-primitive-name-ref"
5510 step
= db_nslcmop_update
[
5512 ] = "executing pre-scale scaling-config-action '{}'".format(
5513 vnf_config_primitive
5516 # look for primitive
5517 for config_primitive
in (
5518 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5519 ).get("config-primitive", ()):
5520 if config_primitive
["name"] == vnf_config_primitive
:
5524 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5525 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5526 "primitive".format(scaling_group
, vnf_config_primitive
)
5529 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5530 if db_vnfr
.get("additionalParamsForVnf"):
5531 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5533 scale_process
= "VCA"
5534 db_nsr_update
["config-status"] = "configuring pre-scaling"
5535 primitive_params
= self
._map
_primitive
_params
(
5536 config_primitive
, {}, vnfr_params
5539 # Pre-scale retry check: Check if this sub-operation has been executed before
5540 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5543 vnf_config_primitive
,
5547 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5548 # Skip sub-operation
5549 result
= "COMPLETED"
5550 result_detail
= "Done"
5553 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5554 vnf_config_primitive
, result
, result_detail
5558 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5559 # New sub-operation: Get index of this sub-operation
5561 len(db_nslcmop
.get("_admin", {}).get("operations"))
5566 + "vnf_config_primitive={} New sub-operation".format(
5567 vnf_config_primitive
5571 # retry: Get registered params for this existing sub-operation
5572 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5575 vnf_index
= op
.get("member_vnf_index")
5576 vnf_config_primitive
= op
.get("primitive")
5577 primitive_params
= op
.get("primitive_params")
5580 + "vnf_config_primitive={} Sub-operation retry".format(
5581 vnf_config_primitive
5584 # Execute the primitive, either with new (first-time) or registered (reintent) args
5585 ee_descriptor_id
= config_primitive
.get(
5586 "execution-environment-ref"
5588 primitive_name
= config_primitive
.get(
5589 "execution-environment-primitive", vnf_config_primitive
5591 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5592 nsr_deployed
["VCA"],
5593 member_vnf_index
=vnf_index
,
5595 vdu_count_index
=None,
5596 ee_descriptor_id
=ee_descriptor_id
,
5598 result
, result_detail
= await self
._ns
_execute
_primitive
(
5607 + "vnf_config_primitive={} Done with result {} {}".format(
5608 vnf_config_primitive
, result
, result_detail
5611 # Update operationState = COMPLETED | FAILED
5612 self
._update
_suboperation
_status
(
5613 db_nslcmop
, op_index
, result
, result_detail
5616 if result
== "FAILED":
5617 raise LcmException(result_detail
)
5618 db_nsr_update
["config-status"] = old_config_status
5619 scale_process
= None
5623 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5626 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5629 # SCALE-IN VCA - BEGIN
5630 if vca_scaling_info
:
5631 step
= db_nslcmop_update
[
5633 ] = "Deleting the execution environments"
5634 scale_process
= "VCA"
5635 for vca_info
in vca_scaling_info
:
5636 if vca_info
["type"] == "delete":
5637 member_vnf_index
= str(vca_info
["member-vnf-index"])
5639 logging_text
+ "vdu info: {}".format(vca_info
)
5641 if vca_info
.get("osm_vdu_id"):
5642 vdu_id
= vca_info
["osm_vdu_id"]
5643 vdu_index
= int(vca_info
["vdu_index"])
5646 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5647 member_vnf_index
, vdu_id
, vdu_index
5651 kdu_id
= vca_info
["osm_kdu_id"]
5654 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5655 member_vnf_index
, kdu_id
, vdu_index
5657 stage
[2] = step
= "Scaling in VCA"
5658 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5659 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5660 config_update
= db_nsr
["configurationStatus"]
5661 for vca_index
, vca
in enumerate(vca_update
):
5663 (vca
or vca
.get("ee_id"))
5664 and vca
["member-vnf-index"] == member_vnf_index
5665 and vca
["vdu_count_index"] == vdu_index
5667 if vca
.get("vdu_id"):
5668 config_descriptor
= get_configuration(
5669 db_vnfd
, vca
.get("vdu_id")
5671 elif vca
.get("kdu_name"):
5672 config_descriptor
= get_configuration(
5673 db_vnfd
, vca
.get("kdu_name")
5676 config_descriptor
= get_configuration(
5677 db_vnfd
, db_vnfd
["id"]
5679 operation_params
= (
5680 db_nslcmop
.get("operationParams") or {}
5682 exec_terminate_primitives
= not operation_params
.get(
5683 "skip_terminate_primitives"
5684 ) and vca
.get("needed_terminate")
5685 task
= asyncio
.ensure_future(
5694 exec_primitives
=exec_terminate_primitives
,
5698 timeout
=self
.timeout_charm_delete
,
5701 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5704 del vca_update
[vca_index
]
5705 del config_update
[vca_index
]
5706 # wait for pending tasks of terminate primitives
5710 + "Waiting for tasks {}".format(
5711 list(tasks_dict_info
.keys())
5714 error_list
= await self
._wait
_for
_tasks
(
5718 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5723 tasks_dict_info
.clear()
5725 raise LcmException("; ".join(error_list
))
5727 db_vca_and_config_update
= {
5728 "_admin.deployed.VCA": vca_update
,
5729 "configurationStatus": config_update
,
5732 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5734 scale_process
= None
5735 # SCALE-IN VCA - END
5738 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5739 scale_process
= "RO"
5740 if self
.ro_config
.get("ng"):
5741 await self
._scale
_ng
_ro
(
5742 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5744 scaling_info
.pop("vdu-create", None)
5745 scaling_info
.pop("vdu-delete", None)
5747 scale_process
= None
5751 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5752 scale_process
= "KDU"
5753 await self
._scale
_kdu
(
5754 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5756 scaling_info
.pop("kdu-create", None)
5757 scaling_info
.pop("kdu-delete", None)
5759 scale_process
= None
5763 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5765 # SCALE-UP VCA - BEGIN
5766 if vca_scaling_info
:
5767 step
= db_nslcmop_update
[
5769 ] = "Creating new execution environments"
5770 scale_process
= "VCA"
5771 for vca_info
in vca_scaling_info
:
5772 if vca_info
["type"] == "create":
5773 member_vnf_index
= str(vca_info
["member-vnf-index"])
5775 logging_text
+ "vdu info: {}".format(vca_info
)
5777 vnfd_id
= db_vnfr
["vnfd-ref"]
5778 if vca_info
.get("osm_vdu_id"):
5779 vdu_index
= int(vca_info
["vdu_index"])
5780 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5781 if db_vnfr
.get("additionalParamsForVnf"):
5782 deploy_params
.update(
5784 db_vnfr
["additionalParamsForVnf"].copy()
5787 descriptor_config
= get_configuration(
5788 db_vnfd
, db_vnfd
["id"]
5790 if descriptor_config
:
5795 logging_text
=logging_text
5796 + "member_vnf_index={} ".format(member_vnf_index
),
5799 nslcmop_id
=nslcmop_id
,
5805 member_vnf_index
=member_vnf_index
,
5806 vdu_index
=vdu_index
,
5808 deploy_params
=deploy_params
,
5809 descriptor_config
=descriptor_config
,
5810 base_folder
=base_folder
,
5811 task_instantiation_info
=tasks_dict_info
,
5814 vdu_id
= vca_info
["osm_vdu_id"]
5815 vdur
= find_in_list(
5816 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5818 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5819 if vdur
.get("additionalParams"):
5820 deploy_params_vdu
= parse_yaml_strings(
5821 vdur
["additionalParams"]
5824 deploy_params_vdu
= deploy_params
5825 deploy_params_vdu
["OSM"] = get_osm_params(
5826 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5828 if descriptor_config
:
5833 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5834 member_vnf_index
, vdu_id
, vdu_index
5836 stage
[2] = step
= "Scaling out VCA"
5837 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5839 logging_text
=logging_text
5840 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5841 member_vnf_index
, vdu_id
, vdu_index
5845 nslcmop_id
=nslcmop_id
,
5851 member_vnf_index
=member_vnf_index
,
5852 vdu_index
=vdu_index
,
5854 deploy_params
=deploy_params_vdu
,
5855 descriptor_config
=descriptor_config
,
5856 base_folder
=base_folder
,
5857 task_instantiation_info
=tasks_dict_info
,
5861 kdu_name
= vca_info
["osm_kdu_id"]
5862 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5863 if descriptor_config
:
5865 kdu_index
= int(vca_info
["kdu_index"])
5869 for x
in db_vnfr
["kdur"]
5870 if x
["kdu-name"] == kdu_name
5872 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5873 if kdur
.get("additionalParams"):
5874 deploy_params_kdu
= parse_yaml_strings(
5875 kdur
["additionalParams"]
5879 logging_text
=logging_text
,
5882 nslcmop_id
=nslcmop_id
,
5888 member_vnf_index
=member_vnf_index
,
5889 vdu_index
=kdu_index
,
5891 deploy_params
=deploy_params_kdu
,
5892 descriptor_config
=descriptor_config
,
5893 base_folder
=base_folder
,
5894 task_instantiation_info
=tasks_dict_info
,
5897 # SCALE-UP VCA - END
5898 scale_process
= None
5901 # execute primitive service POST-SCALING
5902 step
= "Executing post-scale vnf-config-primitive"
5903 if scaling_descriptor
.get("scaling-config-action"):
5904 for scaling_config_action
in scaling_descriptor
[
5905 "scaling-config-action"
5908 scaling_config_action
.get("trigger") == "post-scale-in"
5909 and scaling_type
== "SCALE_IN"
5911 scaling_config_action
.get("trigger") == "post-scale-out"
5912 and scaling_type
== "SCALE_OUT"
5914 vnf_config_primitive
= scaling_config_action
[
5915 "vnf-config-primitive-name-ref"
5917 step
= db_nslcmop_update
[
5919 ] = "executing post-scale scaling-config-action '{}'".format(
5920 vnf_config_primitive
5923 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5924 if db_vnfr
.get("additionalParamsForVnf"):
5925 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5927 # look for primitive
5928 for config_primitive
in (
5929 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5930 ).get("config-primitive", ()):
5931 if config_primitive
["name"] == vnf_config_primitive
:
5935 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
5936 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
5937 "config-primitive".format(
5938 scaling_group
, vnf_config_primitive
5941 scale_process
= "VCA"
5942 db_nsr_update
["config-status"] = "configuring post-scaling"
5943 primitive_params
= self
._map
_primitive
_params
(
5944 config_primitive
, {}, vnfr_params
5947 # Post-scale retry check: Check if this sub-operation has been executed before
5948 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5951 vnf_config_primitive
,
5955 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5956 # Skip sub-operation
5957 result
= "COMPLETED"
5958 result_detail
= "Done"
5961 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5962 vnf_config_primitive
, result
, result_detail
5966 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5967 # New sub-operation: Get index of this sub-operation
5969 len(db_nslcmop
.get("_admin", {}).get("operations"))
5974 + "vnf_config_primitive={} New sub-operation".format(
5975 vnf_config_primitive
5979 # retry: Get registered params for this existing sub-operation
5980 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5983 vnf_index
= op
.get("member_vnf_index")
5984 vnf_config_primitive
= op
.get("primitive")
5985 primitive_params
= op
.get("primitive_params")
5988 + "vnf_config_primitive={} Sub-operation retry".format(
5989 vnf_config_primitive
5992 # Execute the primitive, either with new (first-time) or registered (reintent) args
5993 ee_descriptor_id
= config_primitive
.get(
5994 "execution-environment-ref"
5996 primitive_name
= config_primitive
.get(
5997 "execution-environment-primitive", vnf_config_primitive
5999 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6000 nsr_deployed
["VCA"],
6001 member_vnf_index
=vnf_index
,
6003 vdu_count_index
=None,
6004 ee_descriptor_id
=ee_descriptor_id
,
6006 result
, result_detail
= await self
._ns
_execute
_primitive
(
6015 + "vnf_config_primitive={} Done with result {} {}".format(
6016 vnf_config_primitive
, result
, result_detail
6019 # Update operationState = COMPLETED | FAILED
6020 self
._update
_suboperation
_status
(
6021 db_nslcmop
, op_index
, result
, result_detail
6024 if result
== "FAILED":
6025 raise LcmException(result_detail
)
6026 db_nsr_update
["config-status"] = old_config_status
6027 scale_process
= None
6032 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6033 db_nsr_update
["operational-status"] = (
6035 if old_operational_status
== "failed"
6036 else old_operational_status
6038 db_nsr_update
["config-status"] = old_config_status
6041 ROclient
.ROClientException
,
6046 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6048 except asyncio
.CancelledError
:
6050 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6052 exc
= "Operation was cancelled"
6053 except Exception as e
:
6054 exc
= traceback
.format_exc()
6055 self
.logger
.critical(
6056 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6060 self
._write
_ns
_status
(
6063 current_operation
="IDLE",
6064 current_operation_id
=None,
6067 stage
[1] = "Waiting for instantiate pending tasks."
6068 self
.logger
.debug(logging_text
+ stage
[1])
6069 exc
= await self
._wait
_for
_tasks
(
6072 self
.timeout_ns_deploy
,
6080 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6081 nslcmop_operation_state
= "FAILED"
6083 db_nsr_update
["operational-status"] = old_operational_status
6084 db_nsr_update
["config-status"] = old_config_status
6085 db_nsr_update
["detailed-status"] = ""
6087 if "VCA" in scale_process
:
6088 db_nsr_update
["config-status"] = "failed"
6089 if "RO" in scale_process
:
6090 db_nsr_update
["operational-status"] = "failed"
6093 ] = "FAILED scaling nslcmop={} {}: {}".format(
6094 nslcmop_id
, step
, exc
6097 error_description_nslcmop
= None
6098 nslcmop_operation_state
= "COMPLETED"
6099 db_nslcmop_update
["detailed-status"] = "Done"
6101 self
._write
_op
_status
(
6104 error_message
=error_description_nslcmop
,
6105 operation_state
=nslcmop_operation_state
,
6106 other_update
=db_nslcmop_update
,
6109 self
._write
_ns
_status
(
6112 current_operation
="IDLE",
6113 current_operation_id
=None,
6114 other_update
=db_nsr_update
,
6117 if nslcmop_operation_state
:
6121 "nslcmop_id": nslcmop_id
,
6122 "operationState": nslcmop_operation_state
,
6124 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6125 except Exception as e
:
6127 logging_text
+ "kafka_write notification Exception {}".format(e
)
6129 self
.logger
.debug(logging_text
+ "Exit")
6130 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6132 async def _scale_kdu(
6133 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6135 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6136 for kdu_name
in _scaling_info
:
6137 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6138 deployed_kdu
, index
= get_deployed_kdu(
6139 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6141 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6142 kdu_instance
= deployed_kdu
["kdu-instance"]
6143 scale
= int(kdu_scaling_info
["scale"])
6144 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6147 "collection": "nsrs",
6148 "filter": {"_id": nsr_id
},
6149 "path": "_admin.deployed.K8s.{}".format(index
),
6152 step
= "scaling application {}".format(
6153 kdu_scaling_info
["resource-name"]
6155 self
.logger
.debug(logging_text
+ step
)
6157 if kdu_scaling_info
["type"] == "delete":
6158 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6161 and kdu_config
.get("terminate-config-primitive")
6162 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6164 terminate_config_primitive_list
= kdu_config
.get(
6165 "terminate-config-primitive"
6167 terminate_config_primitive_list
.sort(
6168 key
=lambda val
: int(val
["seq"])
6172 terminate_config_primitive
6173 ) in terminate_config_primitive_list
:
6174 primitive_params_
= self
._map
_primitive
_params
(
6175 terminate_config_primitive
, {}, {}
6177 step
= "execute terminate config primitive"
6178 self
.logger
.debug(logging_text
+ step
)
6179 await asyncio
.wait_for(
6180 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6181 cluster_uuid
=cluster_uuid
,
6182 kdu_instance
=kdu_instance
,
6183 primitive_name
=terminate_config_primitive
["name"],
6184 params
=primitive_params_
,
6191 await asyncio
.wait_for(
6192 self
.k8scluster_map
[k8s_cluster_type
].scale(
6195 kdu_scaling_info
["resource-name"],
6198 timeout
=self
.timeout_vca_on_error
,
6201 if kdu_scaling_info
["type"] == "create":
6202 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6205 and kdu_config
.get("initial-config-primitive")
6206 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6208 initial_config_primitive_list
= kdu_config
.get(
6209 "initial-config-primitive"
6211 initial_config_primitive_list
.sort(
6212 key
=lambda val
: int(val
["seq"])
6215 for initial_config_primitive
in initial_config_primitive_list
:
6216 primitive_params_
= self
._map
_primitive
_params
(
6217 initial_config_primitive
, {}, {}
6219 step
= "execute initial config primitive"
6220 self
.logger
.debug(logging_text
+ step
)
6221 await asyncio
.wait_for(
6222 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6223 cluster_uuid
=cluster_uuid
,
6224 kdu_instance
=kdu_instance
,
6225 primitive_name
=initial_config_primitive
["name"],
6226 params
=primitive_params_
,
6233 async def _scale_ng_ro(
6234 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6236 nsr_id
= db_nslcmop
["nsInstanceId"]
6237 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6240 # read from db: vnfd's for every vnf
6243 # for each vnf in ns, read vnfd
6244 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6245 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6246 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6247 # if we haven't this vnfd, read it from db
6248 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6250 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6251 db_vnfds
.append(vnfd
)
6252 n2vc_key
= self
.n2vc
.get_public_key()
6253 n2vc_key_list
= [n2vc_key
]
6256 vdu_scaling_info
.get("vdu-create"),
6257 vdu_scaling_info
.get("vdu-delete"),
6260 # db_vnfr has been updated, update db_vnfrs to use it
6261 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6262 await self
._instantiate
_ng
_ro
(
6272 start_deploy
=time(),
6273 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6275 if vdu_scaling_info
.get("vdu-delete"):
6277 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6280 async def add_prometheus_metrics(
6281 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6283 if not self
.prometheus
:
6285 # look if exist a file called 'prometheus*.j2' and
6286 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6290 for f
in artifact_content
6291 if f
.startswith("prometheus") and f
.endswith(".j2")
6297 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6301 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6302 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6304 vnfr_id
= vnfr_id
.replace("-", "")
6306 "JOB_NAME": vnfr_id
,
6307 "TARGET_IP": target_ip
,
6308 "EXPORTER_POD_IP": host_name
,
6309 "EXPORTER_POD_PORT": host_port
,
6311 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
6312 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6313 for job
in job_list
:
6315 not isinstance(job
.get("job_name"), str)
6316 or vnfr_id
not in job
["job_name"]
6318 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6319 job
["nsr_id"] = nsr_id
6320 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
6321 if await self
.prometheus
.update(job_dict
):
6322 return list(job_dict
.keys())
6324 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6326 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6328 :param: vim_account_id: VIM Account ID
6330 :return: (cloud_name, cloud_credential)
6332 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6333 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6335 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6337 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6339 :param: vim_account_id: VIM Account ID
6341 :return: (cloud_name, cloud_credential)
6343 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6344 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")