1 # -*- coding: utf-8 -*-
4 # Copyright 2018 Telefonica S.A.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
20 from typing
import Any
, Dict
, List
23 import logging
.handlers
34 from osm_lcm
import ROclient
35 from osm_lcm
.data_utils
.nsr
import (
38 get_deployed_vca_list
,
41 from osm_lcm
.data_utils
.vca
import (
50 from osm_lcm
.ng_ro
import NgRoClient
, NgRoException
51 from osm_lcm
.lcm_utils
import (
59 from osm_lcm
.data_utils
.nsd
import (
60 get_ns_configuration_relation_list
,
64 from osm_lcm
.data_utils
.vnfd
import (
68 get_ee_sorted_initial_config_primitive_list
,
69 get_ee_sorted_terminate_config_primitive_list
,
71 get_virtual_link_profiles
,
76 get_number_of_instances
,
78 get_kdu_resource_profile
,
80 from osm_lcm
.data_utils
.list_utils
import find_in_list
81 from osm_lcm
.data_utils
.vnfr
import get_osm_params
, get_vdur_index
, get_kdur
82 from osm_lcm
.data_utils
.dict_utils
import parse_yaml_strings
83 from osm_lcm
.data_utils
.database
.vim_account
import VimAccountDB
84 from n2vc
.definitions
import RelationEndpoint
85 from n2vc
.k8s_helm_conn
import K8sHelmConnector
86 from n2vc
.k8s_helm3_conn
import K8sHelm3Connector
87 from n2vc
.k8s_juju_conn
import K8sJujuConnector
89 from osm_common
.dbbase
import DbException
90 from osm_common
.fsbase
import FsException
92 from osm_lcm
.data_utils
.database
.database
import Database
93 from osm_lcm
.data_utils
.filesystem
.filesystem
import Filesystem
95 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
96 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
98 from osm_lcm
.lcm_helm_conn
import LCMHelmConn
100 from copy
import copy
, deepcopy
101 from time
import time
102 from uuid
import uuid4
104 from random
import randint
106 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
109 class NsLcm(LcmBase
):
110 timeout_vca_on_error
= (
112 ) # Time for charm from first time at blocked,error status to mark as failed
113 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
114 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
115 timeout_charm_delete
= 10 * 60
116 timeout_primitive
= 30 * 60 # timeout for primitive execution
117 timeout_progress_primitive
= (
119 ) # timeout for some progress in a primitive execution
121 SUBOPERATION_STATUS_NOT_FOUND
= -1
122 SUBOPERATION_STATUS_NEW
= -2
123 SUBOPERATION_STATUS_SKIP
= -3
124 task_name_deploy_vca
= "Deploying VCA"
126 def __init__(self
, msg
, lcm_tasks
, config
, loop
, prometheus
=None):
128 Init, Connect to database, filesystem storage, and messaging
129 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
132 super().__init
__(msg
=msg
, logger
=logging
.getLogger("lcm.ns"))
134 self
.db
= Database().instance
.db
135 self
.fs
= Filesystem().instance
.fs
137 self
.lcm_tasks
= lcm_tasks
138 self
.timeout
= config
["timeout"]
139 self
.ro_config
= config
["ro_config"]
140 self
.ng_ro
= config
["ro_config"].get("ng")
141 self
.vca_config
= config
["VCA"].copy()
143 # create N2VC connector
144 self
.n2vc
= N2VCJujuConnector(
147 on_update_db
=self
._on
_update
_n
2vc
_db
,
152 self
.conn_helm_ee
= LCMHelmConn(
155 vca_config
=self
.vca_config
,
156 on_update_db
=self
._on
_update
_n
2vc
_db
,
159 self
.k8sclusterhelm2
= K8sHelmConnector(
160 kubectl_command
=self
.vca_config
.get("kubectlpath"),
161 helm_command
=self
.vca_config
.get("helmpath"),
168 self
.k8sclusterhelm3
= K8sHelm3Connector(
169 kubectl_command
=self
.vca_config
.get("kubectlpath"),
170 helm_command
=self
.vca_config
.get("helm3path"),
177 self
.k8sclusterjuju
= K8sJujuConnector(
178 kubectl_command
=self
.vca_config
.get("kubectlpath"),
179 juju_command
=self
.vca_config
.get("jujupath"),
182 on_update_db
=self
._on
_update
_k
8s
_db
,
187 self
.k8scluster_map
= {
188 "helm-chart": self
.k8sclusterhelm2
,
189 "helm-chart-v3": self
.k8sclusterhelm3
,
190 "chart": self
.k8sclusterhelm3
,
191 "juju-bundle": self
.k8sclusterjuju
,
192 "juju": self
.k8sclusterjuju
,
196 "lxc_proxy_charm": self
.n2vc
,
197 "native_charm": self
.n2vc
,
198 "k8s_proxy_charm": self
.n2vc
,
199 "helm": self
.conn_helm_ee
,
200 "helm-v3": self
.conn_helm_ee
,
203 self
.prometheus
= prometheus
206 self
.RO
= NgRoClient(self
.loop
, **self
.ro_config
)
209 def increment_ip_mac(ip_mac
, vm_index
=1):
210 if not isinstance(ip_mac
, str):
213 # try with ipv4 look for last dot
214 i
= ip_mac
.rfind(".")
217 return "{}{}".format(ip_mac
[:i
], int(ip_mac
[i
:]) + vm_index
)
218 # try with ipv6 or mac look for last colon. Operate in hex
219 i
= ip_mac
.rfind(":")
222 # format in hex, len can be 2 for mac or 4 for ipv6
223 return ("{}{:0" + str(len(ip_mac
) - i
) + "x}").format(
224 ip_mac
[:i
], int(ip_mac
[i
:], 16) + vm_index
230 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
232 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
235 # TODO filter RO descriptor fields...
239 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
240 db_dict
["deploymentStatus"] = ro_descriptor
241 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
243 except Exception as e
:
245 "Cannot write database RO deployment for ns={} -> {}".format(nsrs_id
, e
)
248 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
, vca_id
=None):
250 # remove last dot from path (if exists)
251 if path
.endswith("."):
254 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
255 # .format(table, filter, path, updated_data))
258 nsr_id
= filter.get("_id")
260 # read ns record from database
261 nsr
= self
.db
.get_one(table
="nsrs", q_filter
=filter)
262 current_ns_status
= nsr
.get("nsState")
264 # get vca status for NS
265 status_dict
= await self
.n2vc
.get_status(
266 namespace
="." + nsr_id
, yaml_format
=False, vca_id
=vca_id
271 db_dict
["vcaStatus"] = status_dict
272 await self
.n2vc
.update_vca_status(db_dict
["vcaStatus"], vca_id
=vca_id
)
274 # update configurationStatus for this VCA
276 vca_index
= int(path
[path
.rfind(".") + 1 :])
279 target_dict
=nsr
, key_list
=("_admin", "deployed", "VCA")
281 vca_status
= vca_list
[vca_index
].get("status")
283 configuration_status_list
= nsr
.get("configurationStatus")
284 config_status
= configuration_status_list
[vca_index
].get("status")
286 if config_status
== "BROKEN" and vca_status
!= "failed":
287 db_dict
["configurationStatus"][vca_index
] = "READY"
288 elif config_status
!= "BROKEN" and vca_status
== "failed":
289 db_dict
["configurationStatus"][vca_index
] = "BROKEN"
290 except Exception as e
:
291 # not update configurationStatus
292 self
.logger
.debug("Error updating vca_index (ignore): {}".format(e
))
294 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
295 # if nsState = 'DEGRADED' check if all is OK
297 if current_ns_status
in ("READY", "DEGRADED"):
298 error_description
= ""
300 if status_dict
.get("machines"):
301 for machine_id
in status_dict
.get("machines"):
302 machine
= status_dict
.get("machines").get(machine_id
)
303 # check machine agent-status
304 if machine
.get("agent-status"):
305 s
= machine
.get("agent-status").get("status")
308 error_description
+= (
309 "machine {} agent-status={} ; ".format(
313 # check machine instance status
314 if machine
.get("instance-status"):
315 s
= machine
.get("instance-status").get("status")
318 error_description
+= (
319 "machine {} instance-status={} ; ".format(
324 if status_dict
.get("applications"):
325 for app_id
in status_dict
.get("applications"):
326 app
= status_dict
.get("applications").get(app_id
)
327 # check application status
328 if app
.get("status"):
329 s
= app
.get("status").get("status")
332 error_description
+= (
333 "application {} status={} ; ".format(app_id
, s
)
336 if error_description
:
337 db_dict
["errorDescription"] = error_description
338 if current_ns_status
== "READY" and is_degraded
:
339 db_dict
["nsState"] = "DEGRADED"
340 if current_ns_status
== "DEGRADED" and not is_degraded
:
341 db_dict
["nsState"] = "READY"
344 self
.update_db_2("nsrs", nsr_id
, db_dict
)
346 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
348 except Exception as e
:
349 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
351 async def _on_update_k8s_db(
352 self
, cluster_uuid
, kdu_instance
, filter=None, vca_id
=None
355 Updating vca status in NSR record
356 :param cluster_uuid: UUID of a k8s cluster
357 :param kdu_instance: The unique name of the KDU instance
358 :param filter: To get nsr_id
362 # self.logger.debug("_on_update_k8s_db(cluster_uuid={}, kdu_instance={}, filter={}"
363 # .format(cluster_uuid, kdu_instance, filter))
366 nsr_id
= filter.get("_id")
368 # get vca status for NS
369 vca_status
= await self
.k8sclusterjuju
.status_kdu(
372 complete_status
=True,
378 db_dict
["vcaStatus"] = {nsr_id
: vca_status
}
380 await self
.k8sclusterjuju
.update_vca_status(
381 db_dict
["vcaStatus"],
387 self
.update_db_2("nsrs", nsr_id
, db_dict
)
389 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
391 except Exception as e
:
392 self
.logger
.warn("Error updating NS state for ns={}: {}".format(nsr_id
, e
))
395 def _parse_cloud_init(cloud_init_text
, additional_params
, vnfd_id
, vdu_id
):
397 env
= Environment(undefined
=StrictUndefined
)
398 template
= env
.from_string(cloud_init_text
)
399 return template
.render(additional_params
or {})
400 except UndefinedError
as e
:
402 "Variable {} at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
403 "file, must be provided in the instantiation parameters inside the "
404 "'additionalParamsForVnf/Vdu' block".format(e
, vnfd_id
, vdu_id
)
406 except (TemplateError
, TemplateNotFound
) as e
:
408 "Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".format(
413 def _get_vdu_cloud_init_content(self
, vdu
, vnfd
):
414 cloud_init_content
= cloud_init_file
= None
416 if vdu
.get("cloud-init-file"):
417 base_folder
= vnfd
["_admin"]["storage"]
418 if base_folder
["pkg-dir"]:
419 cloud_init_file
= "{}/{}/cloud_init/{}".format(
420 base_folder
["folder"],
421 base_folder
["pkg-dir"],
422 vdu
["cloud-init-file"],
425 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
426 base_folder
["folder"],
427 vdu
["cloud-init-file"],
429 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
430 cloud_init_content
= ci_file
.read()
431 elif vdu
.get("cloud-init"):
432 cloud_init_content
= vdu
["cloud-init"]
434 return cloud_init_content
435 except FsException
as e
:
437 "Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".format(
438 vnfd
["id"], vdu
["id"], cloud_init_file
, e
442 def _get_vdu_additional_params(self
, db_vnfr
, vdu_id
):
444 vdur
for vdur
in db_vnfr
.get("vdur") if vdu_id
== vdur
["vdu-id-ref"]
446 additional_params
= vdur
.get("additionalParams")
447 return parse_yaml_strings(additional_params
)
449 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
451 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
452 :param vnfd: input vnfd
453 :param new_id: overrides vnf id if provided
454 :param additionalParams: Instantiation params for VNFs provided
455 :param nsrId: Id of the NSR
456 :return: copy of vnfd
458 vnfd_RO
= deepcopy(vnfd
)
459 # remove unused by RO configuration, monitoring, scaling and internal keys
460 vnfd_RO
.pop("_id", None)
461 vnfd_RO
.pop("_admin", None)
462 vnfd_RO
.pop("monitoring-param", None)
463 vnfd_RO
.pop("scaling-group-descriptor", None)
464 vnfd_RO
.pop("kdu", None)
465 vnfd_RO
.pop("k8s-cluster", None)
467 vnfd_RO
["id"] = new_id
469 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
470 for vdu
in get_iterable(vnfd_RO
, "vdu"):
471 vdu
.pop("cloud-init-file", None)
472 vdu
.pop("cloud-init", None)
476 def ip_profile_2_RO(ip_profile
):
477 RO_ip_profile
= deepcopy(ip_profile
)
478 if "dns-server" in RO_ip_profile
:
479 if isinstance(RO_ip_profile
["dns-server"], list):
480 RO_ip_profile
["dns-address"] = []
481 for ds
in RO_ip_profile
.pop("dns-server"):
482 RO_ip_profile
["dns-address"].append(ds
["address"])
484 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
485 if RO_ip_profile
.get("ip-version") == "ipv4":
486 RO_ip_profile
["ip-version"] = "IPv4"
487 if RO_ip_profile
.get("ip-version") == "ipv6":
488 RO_ip_profile
["ip-version"] = "IPv6"
489 if "dhcp-params" in RO_ip_profile
:
490 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
493 def _get_ro_vim_id_for_vim_account(self
, vim_account
):
494 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
495 if db_vim
["_admin"]["operationalState"] != "ENABLED":
497 "VIM={} is not available. operationalState={}".format(
498 vim_account
, db_vim
["_admin"]["operationalState"]
501 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
504 def get_ro_wim_id_for_wim_account(self
, wim_account
):
505 if isinstance(wim_account
, str):
506 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
507 if db_wim
["_admin"]["operationalState"] != "ENABLED":
509 "WIM={} is not available. operationalState={}".format(
510 wim_account
, db_wim
["_admin"]["operationalState"]
513 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
518 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None, mark_delete
=False):
520 db_vdu_push_list
= []
521 db_update
= {"_admin.modified": time()}
523 for vdu_id
, vdu_count
in vdu_create
.items():
527 for vdur
in reversed(db_vnfr
["vdur"])
528 if vdur
["vdu-id-ref"] == vdu_id
534 "Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
539 for count
in range(vdu_count
):
540 vdur_copy
= deepcopy(vdur
)
541 vdur_copy
["status"] = "BUILD"
542 vdur_copy
["status-detailed"] = None
543 vdur_copy
["ip-address"]: None
544 vdur_copy
["_id"] = str(uuid4())
545 vdur_copy
["count-index"] += count
+ 1
546 vdur_copy
["id"] = "{}-{}".format(
547 vdur_copy
["vdu-id-ref"], vdur_copy
["count-index"]
549 vdur_copy
.pop("vim_info", None)
550 for iface
in vdur_copy
["interfaces"]:
551 if iface
.get("fixed-ip"):
552 iface
["ip-address"] = self
.increment_ip_mac(
553 iface
["ip-address"], count
+ 1
556 iface
.pop("ip-address", None)
557 if iface
.get("fixed-mac"):
558 iface
["mac-address"] = self
.increment_ip_mac(
559 iface
["mac-address"], count
+ 1
562 iface
.pop("mac-address", None)
565 ) # only first vdu can be managment of vnf
566 db_vdu_push_list
.append(vdur_copy
)
567 # self.logger.debug("scale out, adding vdu={}".format(vdur_copy))
569 for vdu_id
, vdu_count
in vdu_delete
.items():
571 indexes_to_delete
= [
573 for iv
in enumerate(db_vnfr
["vdur"])
574 if iv
[1]["vdu-id-ref"] == vdu_id
578 "vdur.{}.status".format(i
): "DELETING"
579 for i
in indexes_to_delete
[-vdu_count
:]
583 # it must be deleted one by one because common.db does not allow otherwise
586 for v
in reversed(db_vnfr
["vdur"])
587 if v
["vdu-id-ref"] == vdu_id
589 for vdu
in vdus_to_delete
[:vdu_count
]:
592 {"_id": db_vnfr
["_id"]},
594 pull
={"vdur": {"_id": vdu
["_id"]}},
596 db_push
= {"vdur": db_vdu_push_list
} if db_vdu_push_list
else None
597 self
.db
.set_one("vnfrs", {"_id": db_vnfr
["_id"]}, db_update
, push_list
=db_push
)
598 # modify passed dictionary db_vnfr
599 db_vnfr_
= self
.db
.get_one("vnfrs", {"_id": db_vnfr
["_id"]})
600 db_vnfr
["vdur"] = db_vnfr_
["vdur"]
602 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
604 Updates database nsr with the RO info for the created vld
605 :param ns_update_nsr: dictionary to be filled with the updated info
606 :param db_nsr: content of db_nsr. This is also modified
607 :param nsr_desc_RO: nsr descriptor from RO
608 :return: Nothing, LcmException is raised on errors
611 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
612 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
613 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
615 vld
["vim-id"] = net_RO
.get("vim_net_id")
616 vld
["name"] = net_RO
.get("vim_name")
617 vld
["status"] = net_RO
.get("status")
618 vld
["status-detailed"] = net_RO
.get("error_msg")
619 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
623 "ns_update_nsr: Not found vld={} at RO info".format(vld
["id"])
626 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
628 for db_vnfr
in db_vnfrs
.values():
629 vnfr_update
= {"status": "ERROR"}
630 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
631 if "status" not in vdur
:
632 vdur
["status"] = "ERROR"
633 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
635 vdur
["status-detailed"] = str(error_text
)
637 "vdur.{}.status-detailed".format(vdu_index
)
639 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
640 except DbException
as e
:
641 self
.logger
.error("Cannot update vnf. {}".format(e
))
643 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
645 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
646 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
647 :param nsr_desc_RO: nsr descriptor from RO
648 :return: Nothing, LcmException is raised on errors
650 for vnf_index
, db_vnfr
in db_vnfrs
.items():
651 for vnf_RO
in nsr_desc_RO
["vnfs"]:
652 if vnf_RO
["member_vnf_index"] != vnf_index
:
655 if vnf_RO
.get("ip_address"):
656 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
[
659 elif not db_vnfr
.get("ip-address"):
660 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
661 raise LcmExceptionNoMgmtIP(
662 "ns member_vnf_index '{}' has no IP address".format(
667 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
668 vdur_RO_count_index
= 0
669 if vdur
.get("pdu-type"):
671 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
672 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
674 if vdur
["count-index"] != vdur_RO_count_index
:
675 vdur_RO_count_index
+= 1
677 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
678 if vdur_RO
.get("ip_address"):
679 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
681 vdur
["ip-address"] = None
682 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
683 vdur
["name"] = vdur_RO
.get("vim_name")
684 vdur
["status"] = vdur_RO
.get("status")
685 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
686 for ifacer
in get_iterable(vdur
, "interfaces"):
687 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
688 if ifacer
["name"] == interface_RO
.get("internal_name"):
689 ifacer
["ip-address"] = interface_RO
.get(
692 ifacer
["mac-address"] = interface_RO
.get(
698 "ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
699 "from VIM info".format(
700 vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]
703 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
707 "ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
709 vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]
713 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
714 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
715 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
717 vld
["vim-id"] = net_RO
.get("vim_net_id")
718 vld
["name"] = net_RO
.get("vim_name")
719 vld
["status"] = net_RO
.get("status")
720 vld
["status-detailed"] = net_RO
.get("error_msg")
721 vnfr_update
["vld.{}".format(vld_index
)] = vld
725 "ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
730 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
735 "ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(
740 def _get_ns_config_info(self
, nsr_id
):
742 Generates a mapping between vnf,vdu elements and the N2VC id
743 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
744 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
745 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
746 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
748 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
749 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
751 ns_config_info
= {"osm-config-mapping": mapping
}
752 for vca
in vca_deployed_list
:
753 if not vca
["member-vnf-index"]:
755 if not vca
["vdu_id"]:
756 mapping
[vca
["member-vnf-index"]] = vca
["application"]
760 vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"]
762 ] = vca
["application"]
763 return ns_config_info
765 async def _instantiate_ng_ro(
782 def get_vim_account(vim_account_id
):
784 if vim_account_id
in db_vims
:
785 return db_vims
[vim_account_id
]
786 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account_id
})
787 db_vims
[vim_account_id
] = db_vim
790 # modify target_vld info with instantiation parameters
791 def parse_vld_instantiation_params(
792 target_vim
, target_vld
, vld_params
, target_sdn
794 if vld_params
.get("ip-profile"):
795 target_vld
["vim_info"][target_vim
]["ip_profile"] = vld_params
[
798 if vld_params
.get("provider-network"):
799 target_vld
["vim_info"][target_vim
]["provider_network"] = vld_params
[
802 if "sdn-ports" in vld_params
["provider-network"] and target_sdn
:
803 target_vld
["vim_info"][target_sdn
]["sdn-ports"] = vld_params
[
806 if vld_params
.get("wimAccountId"):
807 target_wim
= "wim:{}".format(vld_params
["wimAccountId"])
808 target_vld
["vim_info"][target_wim
] = {}
809 for param
in ("vim-network-name", "vim-network-id"):
810 if vld_params
.get(param
):
811 if isinstance(vld_params
[param
], dict):
812 for vim
, vim_net
in vld_params
[param
].items():
813 other_target_vim
= "vim:" + vim
815 target_vld
["vim_info"],
816 (other_target_vim
, param
.replace("-", "_")),
819 else: # isinstance str
820 target_vld
["vim_info"][target_vim
][
821 param
.replace("-", "_")
822 ] = vld_params
[param
]
823 if vld_params
.get("common_id"):
824 target_vld
["common_id"] = vld_params
.get("common_id")
826 nslcmop_id
= db_nslcmop
["_id"]
828 "name": db_nsr
["name"],
831 "image": deepcopy(db_nsr
["image"]),
832 "flavor": deepcopy(db_nsr
["flavor"]),
833 "action_id": nslcmop_id
,
834 "cloud_init_content": {},
836 for image
in target
["image"]:
837 image
["vim_info"] = {}
838 for flavor
in target
["flavor"]:
839 flavor
["vim_info"] = {}
841 if db_nslcmop
.get("lcmOperationType") != "instantiate":
842 # get parameters of instantiation:
843 db_nslcmop_instantiate
= self
.db
.get_list(
846 "nsInstanceId": db_nslcmop
["nsInstanceId"],
847 "lcmOperationType": "instantiate",
850 ns_params
= db_nslcmop_instantiate
.get("operationParams")
852 ns_params
= db_nslcmop
.get("operationParams")
853 ssh_keys_instantiation
= ns_params
.get("ssh_keys") or []
854 ssh_keys_all
= ssh_keys_instantiation
+ (n2vc_key_list
or [])
857 for vld_index
, vld
in enumerate(db_nsr
.get("vld")):
858 target_vim
= "vim:{}".format(ns_params
["vimAccountId"])
862 "mgmt-network": vld
.get("mgmt-network", False),
863 "type": vld
.get("type"),
866 "vim_network_name": vld
.get("vim-network-name"),
867 "vim_account_id": ns_params
["vimAccountId"],
871 # check if this network needs SDN assist
872 if vld
.get("pci-interfaces"):
873 db_vim
= get_vim_account(ns_params
["vimAccountId"])
874 sdnc_id
= db_vim
["config"].get("sdn-controller")
876 sdn_vld
= "nsrs:{}:vld.{}".format(nsr_id
, vld
["id"])
877 target_sdn
= "sdn:{}".format(sdnc_id
)
878 target_vld
["vim_info"][target_sdn
] = {
880 "target_vim": target_vim
,
882 "type": vld
.get("type"),
885 nsd_vnf_profiles
= get_vnf_profiles(nsd
)
886 for nsd_vnf_profile
in nsd_vnf_profiles
:
887 for cp
in nsd_vnf_profile
["virtual-link-connectivity"]:
888 if cp
["virtual-link-profile-id"] == vld
["id"]:
890 "member_vnf:{}.{}".format(
891 cp
["constituent-cpd-id"][0][
892 "constituent-base-element-id"
894 cp
["constituent-cpd-id"][0]["constituent-cpd-id"],
896 ] = "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
898 # check at nsd descriptor, if there is an ip-profile
900 nsd_vlp
= find_in_list(
901 get_virtual_link_profiles(nsd
),
902 lambda a_link_profile
: a_link_profile
["virtual-link-desc-id"]
907 and nsd_vlp
.get("virtual-link-protocol-data")
908 and nsd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
910 ip_profile_source_data
= nsd_vlp
["virtual-link-protocol-data"][
913 ip_profile_dest_data
= {}
914 if "ip-version" in ip_profile_source_data
:
915 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
918 if "cidr" in ip_profile_source_data
:
919 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
922 if "gateway-ip" in ip_profile_source_data
:
923 ip_profile_dest_data
["gateway-address"] = ip_profile_source_data
[
926 if "dhcp-enabled" in ip_profile_source_data
:
927 ip_profile_dest_data
["dhcp-params"] = {
928 "enabled": ip_profile_source_data
["dhcp-enabled"]
930 vld_params
["ip-profile"] = ip_profile_dest_data
932 # update vld_params with instantiation params
933 vld_instantiation_params
= find_in_list(
934 get_iterable(ns_params
, "vld"),
935 lambda a_vld
: a_vld
["name"] in (vld
["name"], vld
["id"]),
937 if vld_instantiation_params
:
938 vld_params
.update(vld_instantiation_params
)
939 parse_vld_instantiation_params(target_vim
, target_vld
, vld_params
, None)
940 target
["ns"]["vld"].append(target_vld
)
942 for vnfr
in db_vnfrs
.values():
944 db_vnfds
, lambda db_vnf
: db_vnf
["id"] == vnfr
["vnfd-ref"]
946 vnf_params
= find_in_list(
947 get_iterable(ns_params
, "vnf"),
948 lambda a_vnf
: a_vnf
["member-vnf-index"] == vnfr
["member-vnf-index-ref"],
950 target_vnf
= deepcopy(vnfr
)
951 target_vim
= "vim:{}".format(vnfr
["vim-account-id"])
952 for vld
in target_vnf
.get("vld", ()):
953 # check if connected to a ns.vld, to fill target'
954 vnf_cp
= find_in_list(
955 vnfd
.get("int-virtual-link-desc", ()),
956 lambda cpd
: cpd
.get("id") == vld
["id"],
959 ns_cp
= "member_vnf:{}.{}".format(
960 vnfr
["member-vnf-index-ref"], vnf_cp
["id"]
962 if cp2target
.get(ns_cp
):
963 vld
["target"] = cp2target
[ns_cp
]
966 target_vim
: {"vim_network_name": vld
.get("vim-network-name")}
968 # check if this network needs SDN assist
970 if vld
.get("pci-interfaces"):
971 db_vim
= get_vim_account(vnfr
["vim-account-id"])
972 sdnc_id
= db_vim
["config"].get("sdn-controller")
974 sdn_vld
= "vnfrs:{}:vld.{}".format(target_vnf
["_id"], vld
["id"])
975 target_sdn
= "sdn:{}".format(sdnc_id
)
976 vld
["vim_info"][target_sdn
] = {
978 "target_vim": target_vim
,
980 "type": vld
.get("type"),
983 # check at vnfd descriptor, if there is an ip-profile
985 vnfd_vlp
= find_in_list(
986 get_virtual_link_profiles(vnfd
),
987 lambda a_link_profile
: a_link_profile
["id"] == vld
["id"],
991 and vnfd_vlp
.get("virtual-link-protocol-data")
992 and vnfd_vlp
["virtual-link-protocol-data"].get("l3-protocol-data")
994 ip_profile_source_data
= vnfd_vlp
["virtual-link-protocol-data"][
997 ip_profile_dest_data
= {}
998 if "ip-version" in ip_profile_source_data
:
999 ip_profile_dest_data
["ip-version"] = ip_profile_source_data
[
1002 if "cidr" in ip_profile_source_data
:
1003 ip_profile_dest_data
["subnet-address"] = ip_profile_source_data
[
1006 if "gateway-ip" in ip_profile_source_data
:
1007 ip_profile_dest_data
[
1009 ] = ip_profile_source_data
["gateway-ip"]
1010 if "dhcp-enabled" in ip_profile_source_data
:
1011 ip_profile_dest_data
["dhcp-params"] = {
1012 "enabled": ip_profile_source_data
["dhcp-enabled"]
1015 vld_params
["ip-profile"] = ip_profile_dest_data
1016 # update vld_params with instantiation params
1018 vld_instantiation_params
= find_in_list(
1019 get_iterable(vnf_params
, "internal-vld"),
1020 lambda i_vld
: i_vld
["name"] == vld
["id"],
1022 if vld_instantiation_params
:
1023 vld_params
.update(vld_instantiation_params
)
1024 parse_vld_instantiation_params(target_vim
, vld
, vld_params
, target_sdn
)
1027 for vdur
in target_vnf
.get("vdur", ()):
1028 if vdur
.get("status") == "DELETING" or vdur
.get("pdu-type"):
1029 continue # This vdu must not be created
1030 vdur
["vim_info"] = {"vim_account_id": vnfr
["vim-account-id"]}
1032 self
.logger
.debug("NS > ssh_keys > {}".format(ssh_keys_all
))
1035 vdu_configuration
= get_configuration(vnfd
, vdur
["vdu-id-ref"])
1036 vnf_configuration
= get_configuration(vnfd
, vnfd
["id"])
1039 and vdu_configuration
.get("config-access")
1040 and vdu_configuration
.get("config-access").get("ssh-access")
1042 vdur
["ssh-keys"] = ssh_keys_all
1043 vdur
["ssh-access-required"] = vdu_configuration
[
1045 ]["ssh-access"]["required"]
1048 and vnf_configuration
.get("config-access")
1049 and vnf_configuration
.get("config-access").get("ssh-access")
1050 and any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"])
1052 vdur
["ssh-keys"] = ssh_keys_all
1053 vdur
["ssh-access-required"] = vnf_configuration
[
1055 ]["ssh-access"]["required"]
1056 elif ssh_keys_instantiation
and find_in_list(
1057 vdur
["interfaces"], lambda iface
: iface
.get("mgmt-vnf")
1059 vdur
["ssh-keys"] = ssh_keys_instantiation
1061 self
.logger
.debug("NS > vdur > {}".format(vdur
))
1063 vdud
= get_vdu(vnfd
, vdur
["vdu-id-ref"])
1065 if vdud
.get("cloud-init-file"):
1066 vdur
["cloud-init"] = "{}:file:{}".format(
1067 vnfd
["_id"], vdud
.get("cloud-init-file")
1069 # read file and put content at target.cloul_init_content. Avoid ng_ro to use shared package system
1070 if vdur
["cloud-init"] not in target
["cloud_init_content"]:
1071 base_folder
= vnfd
["_admin"]["storage"]
1072 if base_folder
["pkg-dir"]:
1073 cloud_init_file
= "{}/{}/cloud_init/{}".format(
1074 base_folder
["folder"],
1075 base_folder
["pkg-dir"],
1076 vdud
.get("cloud-init-file"),
1079 cloud_init_file
= "{}/Scripts/cloud_init/{}".format(
1080 base_folder
["folder"],
1081 vdud
.get("cloud-init-file"),
1083 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
1084 target
["cloud_init_content"][
1087 elif vdud
.get("cloud-init"):
1088 vdur
["cloud-init"] = "{}:vdu:{}".format(
1089 vnfd
["_id"], get_vdu_index(vnfd
, vdur
["vdu-id-ref"])
1091 # put content at target.cloul_init_content. Avoid ng_ro read vnfd descriptor
1092 target
["cloud_init_content"][vdur
["cloud-init"]] = vdud
[
1095 vdur
["additionalParams"] = vdur
.get("additionalParams") or {}
1096 deploy_params_vdu
= self
._format
_additional
_params
(
1097 vdur
.get("additionalParams") or {}
1099 deploy_params_vdu
["OSM"] = get_osm_params(
1100 vnfr
, vdur
["vdu-id-ref"], vdur
["count-index"]
1102 vdur
["additionalParams"] = deploy_params_vdu
1105 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
1106 if target_vim
not in ns_flavor
["vim_info"]:
1107 ns_flavor
["vim_info"][target_vim
] = {}
1110 # in case alternative images are provided we must check if they should be applied
1111 # for the vim_type, modify the vim_type taking into account
1112 ns_image_id
= int(vdur
["ns-image-id"])
1113 if vdur
.get("alt-image-ids"):
1114 db_vim
= get_vim_account(vnfr
["vim-account-id"])
1115 vim_type
= db_vim
["vim_type"]
1116 for alt_image_id
in vdur
.get("alt-image-ids"):
1117 ns_alt_image
= target
["image"][int(alt_image_id
)]
1118 if vim_type
== ns_alt_image
.get("vim-type"):
1119 # must use alternative image
1121 "use alternative image id: {}".format(alt_image_id
)
1123 ns_image_id
= alt_image_id
1124 vdur
["ns-image-id"] = ns_image_id
1126 ns_image
= target
["image"][int(ns_image_id
)]
1127 if target_vim
not in ns_image
["vim_info"]:
1128 ns_image
["vim_info"][target_vim
] = {}
1130 vdur
["vim_info"] = {target_vim
: {}}
1131 # instantiation parameters
1133 # vdu_instantiation_params = next((v for v in get_iterable(vnf_params, "vdu") if v["id"] ==
1134 # vdud["id"]), None)
1135 vdur_list
.append(vdur
)
1136 target_vnf
["vdur"] = vdur_list
1137 target
["vnf"].append(target_vnf
)
1139 desc
= await self
.RO
.deploy(nsr_id
, target
)
1140 self
.logger
.debug("RO return > {}".format(desc
))
1141 action_id
= desc
["action_id"]
1142 await self
._wait
_ng
_ro
(
1143 nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
1148 "_admin.deployed.RO.operational-status": "running",
1149 "detailed-status": " ".join(stage
),
1151 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1152 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1153 self
._write
_op
_status
(nslcmop_id
, stage
)
1155 logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
)
1159 async def _wait_ng_ro(
1168 detailed_status_old
= None
1170 start_time
= start_time
or time()
1171 while time() <= start_time
+ timeout
:
1172 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
1173 self
.logger
.debug("Wait NG RO > {}".format(desc_status
))
1174 if desc_status
["status"] == "FAILED":
1175 raise NgRoException(desc_status
["details"])
1176 elif desc_status
["status"] == "BUILD":
1178 stage
[2] = "VIM: ({})".format(desc_status
["details"])
1179 elif desc_status
["status"] == "DONE":
1181 stage
[2] = "Deployed at VIM"
1184 assert False, "ROclient.check_ns_status returns unknown {}".format(
1185 desc_status
["status"]
1187 if stage
and nslcmop_id
and stage
[2] != detailed_status_old
:
1188 detailed_status_old
= stage
[2]
1189 db_nsr_update
["detailed-status"] = " ".join(stage
)
1190 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1191 self
._write
_op
_status
(nslcmop_id
, stage
)
1192 await asyncio
.sleep(15, loop
=self
.loop
)
1193 else: # timeout_ns_deploy
1194 raise NgRoException("Timeout waiting ns to deploy")
1196 async def _terminate_ng_ro(
1197 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
1202 start_deploy
= time()
1209 "action_id": nslcmop_id
,
1211 desc
= await self
.RO
.deploy(nsr_id
, target
)
1212 action_id
= desc
["action_id"]
1213 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1214 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1217 + "ns terminate action at RO. action_id={}".format(action_id
)
1221 delete_timeout
= 20 * 60 # 20 minutes
1222 await self
._wait
_ng
_ro
(
1223 nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
1226 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1227 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1229 await self
.RO
.delete(nsr_id
)
1230 except Exception as e
:
1231 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1232 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1233 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1234 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1236 logging_text
+ "RO_action_id={} already deleted".format(action_id
)
1238 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1239 failed_detail
.append("delete conflict: {}".format(e
))
1242 + "RO_action_id={} delete conflict: {}".format(action_id
, e
)
1245 failed_detail
.append("delete error: {}".format(e
))
1248 + "RO_action_id={} delete error: {}".format(action_id
, e
)
1252 stage
[2] = "Error deleting from VIM"
1254 stage
[2] = "Deleted from VIM"
1255 db_nsr_update
["detailed-status"] = " ".join(stage
)
1256 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1257 self
._write
_op
_status
(nslcmop_id
, stage
)
1260 raise LcmException("; ".join(failed_detail
))
1263 async def instantiate_RO(
1277 :param logging_text: preffix text to use at logging
1278 :param nsr_id: nsr identity
1279 :param nsd: database content of ns descriptor
1280 :param db_nsr: database content of ns record
1281 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1283 :param db_vnfds: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1284 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1285 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1286 :return: None or exception
1289 start_deploy
= time()
1290 ns_params
= db_nslcmop
.get("operationParams")
1291 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1292 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1294 timeout_ns_deploy
= self
.timeout
.get(
1295 "ns_deploy", self
.timeout_ns_deploy
1298 # Check for and optionally request placement optimization. Database will be updated if placement activated
1299 stage
[2] = "Waiting for Placement."
1300 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1301 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1302 for vnfr
in db_vnfrs
.values():
1303 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1306 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1308 return await self
._instantiate
_ng
_ro
(
1321 except Exception as e
:
1322 stage
[2] = "ERROR deploying at VIM"
1323 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1325 "Error deploying at VIM {}".format(e
),
1326 exc_info
=not isinstance(
1329 ROclient
.ROClientException
,
1338 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1340 Wait for kdu to be up, get ip address
1341 :param logging_text: prefix use for logging
1348 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1351 while nb_tries
< 360:
1352 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1356 for x
in get_iterable(db_vnfr
, "kdur")
1357 if x
.get("kdu-name") == kdu_name
1363 "Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
)
1365 if kdur
.get("status"):
1366 if kdur
["status"] in ("READY", "ENABLED"):
1367 return kdur
.get("ip-address")
1370 "target KDU={} is in error state".format(kdu_name
)
1373 await asyncio
.sleep(10, loop
=self
.loop
)
1375 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1377 async def wait_vm_up_insert_key_ro(
1378 self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None
1381 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1382 :param logging_text: prefix use for logging
1387 :param pub_key: public ssh key to inject, None to skip
1388 :param user: user to apply the public ssh key
1392 self
.logger
.debug(logging_text
+ "Starting wait_vm_up_insert_key_ro")
1396 target_vdu_id
= None
1402 if ro_retries
>= 360: # 1 hour
1404 "Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
)
1407 await asyncio
.sleep(10, loop
=self
.loop
)
1410 if not target_vdu_id
:
1411 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1413 if not vdu_id
: # for the VNF case
1414 if db_vnfr
.get("status") == "ERROR":
1416 "Cannot inject ssh-key because target VNF is in error state"
1418 ip_address
= db_vnfr
.get("ip-address")
1424 for x
in get_iterable(db_vnfr
, "vdur")
1425 if x
.get("ip-address") == ip_address
1433 for x
in get_iterable(db_vnfr
, "vdur")
1434 if x
.get("vdu-id-ref") == vdu_id
1435 and x
.get("count-index") == vdu_index
1441 not vdur
and len(db_vnfr
.get("vdur", ())) == 1
1442 ): # If only one, this should be the target vdu
1443 vdur
= db_vnfr
["vdur"][0]
1446 "Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(
1447 vnfr_id
, vdu_id
, vdu_index
1450 # New generation RO stores information at "vim_info"
1453 if vdur
.get("vim_info"):
1455 t
for t
in vdur
["vim_info"]
1456 ) # there should be only one key
1457 ng_ro_status
= vdur
["vim_info"][target_vim
].get("vim_status")
1459 vdur
.get("pdu-type")
1460 or vdur
.get("status") == "ACTIVE"
1461 or ng_ro_status
== "ACTIVE"
1463 ip_address
= vdur
.get("ip-address")
1466 target_vdu_id
= vdur
["vdu-id-ref"]
1467 elif vdur
.get("status") == "ERROR" or ng_ro_status
== "ERROR":
1469 "Cannot inject ssh-key because target VM is in error state"
1472 if not target_vdu_id
:
1475 # inject public key into machine
1476 if pub_key
and user
:
1477 self
.logger
.debug(logging_text
+ "Inserting RO key")
1478 self
.logger
.debug("SSH > PubKey > {}".format(pub_key
))
1479 if vdur
.get("pdu-type"):
1480 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1483 ro_vm_id
= "{}-{}".format(
1484 db_vnfr
["member-vnf-index-ref"], target_vdu_id
1485 ) # TODO add vdu_index
1489 "action": "inject_ssh_key",
1493 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdur
["id"]}]}],
1495 desc
= await self
.RO
.deploy(nsr_id
, target
)
1496 action_id
= desc
["action_id"]
1497 await self
._wait
_ng
_ro
(nsr_id
, action_id
, timeout
=600)
1500 # wait until NS is deployed at RO
1502 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1503 ro_nsr_id
= deep_get(
1504 db_nsrs
, ("_admin", "deployed", "RO", "nsr_id")
1508 result_dict
= await self
.RO
.create_action(
1510 item_id_name
=ro_nsr_id
,
1512 "add_public_key": pub_key
,
1517 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1518 if not result_dict
or not isinstance(result_dict
, dict):
1520 "Unknown response from RO when injecting key"
1522 for result
in result_dict
.values():
1523 if result
.get("vim_result") == 200:
1526 raise ROclient
.ROClientException(
1527 "error injecting key: {}".format(
1528 result
.get("description")
1532 except NgRoException
as e
:
1534 "Reaching max tries injecting key. Error: {}".format(e
)
1536 except ROclient
.ROClientException
as e
:
1540 + "error injecting key: {}. Retrying until {} seconds".format(
1547 "Reaching max tries injecting key. Error: {}".format(e
)
1554 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1556 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1558 my_vca
= vca_deployed_list
[vca_index
]
1559 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1560 # vdu or kdu: no dependencies
1564 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1565 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1566 configuration_status_list
= db_nsr
["configurationStatus"]
1567 for index
, vca_deployed
in enumerate(configuration_status_list
):
1568 if index
== vca_index
:
1571 if not my_vca
.get("member-vnf-index") or (
1572 vca_deployed
.get("member-vnf-index")
1573 == my_vca
.get("member-vnf-index")
1575 internal_status
= configuration_status_list
[index
].get("status")
1576 if internal_status
== "READY":
1578 elif internal_status
== "BROKEN":
1580 "Configuration aborted because dependent charm/s has failed"
1585 # no dependencies, return
1587 await asyncio
.sleep(10)
1590 raise LcmException("Configuration aborted because dependent charm/s timeout")
1592 def get_vca_id(self
, db_vnfr
: dict, db_nsr
: dict):
1595 vca_id
= deep_get(db_vnfr
, ("vca-id",))
1597 vim_account_id
= deep_get(db_nsr
, ("instantiate_params", "vimAccountId"))
1598 vca_id
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("vca")
1601 async def instantiate_N2VC(
1618 ee_config_descriptor
,
1620 nsr_id
= db_nsr
["_id"]
1621 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1622 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1623 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1624 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1626 "collection": "nsrs",
1627 "filter": {"_id": nsr_id
},
1628 "path": db_update_entry
,
1634 element_under_configuration
= nsr_id
1638 vnfr_id
= db_vnfr
["_id"]
1639 osm_config
["osm"]["vnf_id"] = vnfr_id
1641 namespace
= "{nsi}.{ns}".format(nsi
=nsi_id
if nsi_id
else "", ns
=nsr_id
)
1643 if vca_type
== "native_charm":
1646 index_number
= vdu_index
or 0
1649 element_type
= "VNF"
1650 element_under_configuration
= vnfr_id
1651 namespace
+= ".{}-{}".format(vnfr_id
, index_number
)
1653 namespace
+= ".{}-{}".format(vdu_id
, index_number
)
1654 element_type
= "VDU"
1655 element_under_configuration
= "{}-{}".format(vdu_id
, index_number
)
1656 osm_config
["osm"]["vdu_id"] = vdu_id
1658 namespace
+= ".{}".format(kdu_name
)
1659 element_type
= "KDU"
1660 element_under_configuration
= kdu_name
1661 osm_config
["osm"]["kdu_name"] = kdu_name
1664 if base_folder
["pkg-dir"]:
1665 artifact_path
= "{}/{}/{}/{}".format(
1666 base_folder
["folder"],
1667 base_folder
["pkg-dir"],
1669 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1674 artifact_path
= "{}/Scripts/{}/{}/".format(
1675 base_folder
["folder"],
1677 if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm")
1682 self
.logger
.debug("Artifact path > {}".format(artifact_path
))
1684 # get initial_config_primitive_list that applies to this element
1685 initial_config_primitive_list
= config_descriptor
.get(
1686 "initial-config-primitive"
1690 "Initial config primitive list > {}".format(
1691 initial_config_primitive_list
1695 # add config if not present for NS charm
1696 ee_descriptor_id
= ee_config_descriptor
.get("id")
1697 self
.logger
.debug("EE Descriptor > {}".format(ee_descriptor_id
))
1698 initial_config_primitive_list
= get_ee_sorted_initial_config_primitive_list(
1699 initial_config_primitive_list
, vca_deployed
, ee_descriptor_id
1703 "Initial config primitive list #2 > {}".format(
1704 initial_config_primitive_list
1707 # n2vc_redesign STEP 3.1
1708 # find old ee_id if exists
1709 ee_id
= vca_deployed
.get("ee_id")
1711 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
1712 # create or register execution environment in VCA
1713 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm", "helm-v3"):
1715 self
._write
_configuration
_status
(
1717 vca_index
=vca_index
,
1719 element_under_configuration
=element_under_configuration
,
1720 element_type
=element_type
,
1723 step
= "create execution environment"
1724 self
.logger
.debug(logging_text
+ step
)
1728 if vca_type
== "k8s_proxy_charm":
1729 ee_id
= await self
.vca_map
[vca_type
].install_k8s_proxy_charm(
1730 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1 :],
1731 namespace
=namespace
,
1732 artifact_path
=artifact_path
,
1736 elif vca_type
== "helm" or vca_type
== "helm-v3":
1737 ee_id
, credentials
= await self
.vca_map
[
1739 ].create_execution_environment(
1740 namespace
=namespace
,
1744 artifact_path
=artifact_path
,
1748 ee_id
, credentials
= await self
.vca_map
[
1750 ].create_execution_environment(
1751 namespace
=namespace
,
1757 elif vca_type
== "native_charm":
1758 step
= "Waiting to VM being up and getting IP address"
1759 self
.logger
.debug(logging_text
+ step
)
1760 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1769 credentials
= {"hostname": rw_mgmt_ip
}
1771 username
= deep_get(
1772 config_descriptor
, ("config-access", "ssh-access", "default-user")
1774 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1775 # merged. Meanwhile let's get username from initial-config-primitive
1776 if not username
and initial_config_primitive_list
:
1777 for config_primitive
in initial_config_primitive_list
:
1778 for param
in config_primitive
.get("parameter", ()):
1779 if param
["name"] == "ssh-username":
1780 username
= param
["value"]
1784 "Cannot determine the username neither with 'initial-config-primitive' nor with "
1785 "'config-access.ssh-access.default-user'"
1787 credentials
["username"] = username
1788 # n2vc_redesign STEP 3.2
1790 self
._write
_configuration
_status
(
1792 vca_index
=vca_index
,
1793 status
="REGISTERING",
1794 element_under_configuration
=element_under_configuration
,
1795 element_type
=element_type
,
1798 step
= "register execution environment {}".format(credentials
)
1799 self
.logger
.debug(logging_text
+ step
)
1800 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1801 credentials
=credentials
,
1802 namespace
=namespace
,
1807 # for compatibility with MON/POL modules, the need model and application name at database
1808 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1809 ee_id_parts
= ee_id
.split(".")
1810 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1811 if len(ee_id_parts
) >= 2:
1812 model_name
= ee_id_parts
[0]
1813 application_name
= ee_id_parts
[1]
1814 db_nsr_update
[db_update_entry
+ "model"] = model_name
1815 db_nsr_update
[db_update_entry
+ "application"] = application_name
1817 # n2vc_redesign STEP 3.3
1818 step
= "Install configuration Software"
1820 self
._write
_configuration
_status
(
1822 vca_index
=vca_index
,
1823 status
="INSTALLING SW",
1824 element_under_configuration
=element_under_configuration
,
1825 element_type
=element_type
,
1826 other_update
=db_nsr_update
,
1829 # TODO check if already done
1830 self
.logger
.debug(logging_text
+ step
)
1832 if vca_type
== "native_charm":
1833 config_primitive
= next(
1834 (p
for p
in initial_config_primitive_list
if p
["name"] == "config"),
1837 if config_primitive
:
1838 config
= self
._map
_primitive
_params
(
1839 config_primitive
, {}, deploy_params
1842 if vca_type
== "lxc_proxy_charm":
1843 if element_type
== "NS":
1844 num_units
= db_nsr
.get("config-units") or 1
1845 elif element_type
== "VNF":
1846 num_units
= db_vnfr
.get("config-units") or 1
1847 elif element_type
== "VDU":
1848 for v
in db_vnfr
["vdur"]:
1849 if vdu_id
== v
["vdu-id-ref"]:
1850 num_units
= v
.get("config-units") or 1
1852 if vca_type
!= "k8s_proxy_charm":
1853 await self
.vca_map
[vca_type
].install_configuration_sw(
1855 artifact_path
=artifact_path
,
1858 num_units
=num_units
,
1863 # write in db flag of configuration_sw already installed
1865 "nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True}
1868 # add relations for this VCA (wait for other peers related with this VCA)
1869 await self
._add
_vca
_relations
(
1870 logging_text
=logging_text
,
1873 vca_index
=vca_index
,
1876 # if SSH access is required, then get execution environment SSH public
1877 # if native charm we have waited already to VM be UP
1878 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm", "helm-v3"):
1881 # self.logger.debug("get ssh key block")
1883 config_descriptor
, ("config-access", "ssh-access", "required")
1885 # self.logger.debug("ssh key needed")
1886 # Needed to inject a ssh key
1889 ("config-access", "ssh-access", "default-user"),
1891 step
= "Install configuration Software, getting public ssh key"
1892 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(
1893 ee_id
=ee_id
, db_dict
=db_dict
, vca_id
=vca_id
1896 step
= "Insert public key into VM user={} ssh_key={}".format(
1900 # self.logger.debug("no need to get ssh key")
1901 step
= "Waiting to VM being up and getting IP address"
1902 self
.logger
.debug(logging_text
+ step
)
1904 # n2vc_redesign STEP 5.1
1905 # wait for RO (ip-address) Insert pub_key into VM
1908 rw_mgmt_ip
= await self
.wait_kdu_up(
1909 logging_text
, nsr_id
, vnfr_id
, kdu_name
1912 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(
1922 rw_mgmt_ip
= None # This is for a NS configuration
1924 self
.logger
.debug(logging_text
+ " VM_ip_address={}".format(rw_mgmt_ip
))
1926 # store rw_mgmt_ip in deploy params for later replacement
1927 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1929 # n2vc_redesign STEP 6 Execute initial config primitive
1930 step
= "execute initial config primitive"
1932 # wait for dependent primitives execution (NS -> VNF -> VDU)
1933 if initial_config_primitive_list
:
1934 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1936 # stage, in function of element type: vdu, kdu, vnf or ns
1937 my_vca
= vca_deployed_list
[vca_index
]
1938 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1940 stage
[0] = "Stage 3/5: running Day-1 primitives for VDU."
1941 elif my_vca
.get("member-vnf-index"):
1943 stage
[0] = "Stage 4/5: running Day-1 primitives for VNF."
1946 stage
[0] = "Stage 5/5: running Day-1 primitives for NS."
1948 self
._write
_configuration
_status
(
1949 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="EXECUTING PRIMITIVE"
1952 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
1954 check_if_terminated_needed
= True
1955 for initial_config_primitive
in initial_config_primitive_list
:
1956 # adding information on the vca_deployed if it is a NS execution environment
1957 if not vca_deployed
["member-vnf-index"]:
1958 deploy_params
["ns_config_info"] = json
.dumps(
1959 self
._get
_ns
_config
_info
(nsr_id
)
1961 # TODO check if already done
1962 primitive_params_
= self
._map
_primitive
_params
(
1963 initial_config_primitive
, {}, deploy_params
1966 step
= "execute primitive '{}' params '{}'".format(
1967 initial_config_primitive
["name"], primitive_params_
1969 self
.logger
.debug(logging_text
+ step
)
1970 await self
.vca_map
[vca_type
].exec_primitive(
1972 primitive_name
=initial_config_primitive
["name"],
1973 params_dict
=primitive_params_
,
1978 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1979 if check_if_terminated_needed
:
1980 if config_descriptor
.get("terminate-config-primitive"):
1982 "nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True}
1984 check_if_terminated_needed
= False
1986 # TODO register in database that primitive is done
1988 # STEP 7 Configure metrics
1989 if vca_type
== "helm" or vca_type
== "helm-v3":
1990 prometheus_jobs
= await self
.add_prometheus_metrics(
1992 artifact_path
=artifact_path
,
1993 ee_config_descriptor
=ee_config_descriptor
,
1996 target_ip
=rw_mgmt_ip
,
2002 {db_update_entry
+ "prometheus_jobs": prometheus_jobs
},
2005 step
= "instantiated at VCA"
2006 self
.logger
.debug(logging_text
+ step
)
2008 self
._write
_configuration
_status
(
2009 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="READY"
2012 except Exception as e
: # TODO not use Exception but N2VC exception
2013 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
2015 e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)
2018 "Exception while {} : {}".format(step
, e
), exc_info
=True
2020 self
._write
_configuration
_status
(
2021 nsr_id
=nsr_id
, vca_index
=vca_index
, status
="BROKEN"
2023 raise LcmException("{} {}".format(step
, e
)) from e
2025 def _write_ns_status(
2029 current_operation
: str,
2030 current_operation_id
: str,
2031 error_description
: str = None,
2032 error_detail
: str = None,
2033 other_update
: dict = None,
2036 Update db_nsr fields.
2039 :param current_operation:
2040 :param current_operation_id:
2041 :param error_description:
2042 :param error_detail:
2043 :param other_update: Other required changes at database if provided, will be cleared
2047 db_dict
= other_update
or {}
2050 ] = current_operation_id
# for backward compatibility
2051 db_dict
["_admin.current-operation"] = current_operation_id
2052 db_dict
["_admin.operation-type"] = (
2053 current_operation
if current_operation
!= "IDLE" else None
2055 db_dict
["currentOperation"] = current_operation
2056 db_dict
["currentOperationID"] = current_operation_id
2057 db_dict
["errorDescription"] = error_description
2058 db_dict
["errorDetail"] = error_detail
2061 db_dict
["nsState"] = ns_state
2062 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2063 except DbException
as e
:
2064 self
.logger
.warn("Error writing NS status, ns={}: {}".format(nsr_id
, e
))
2066 def _write_op_status(
2070 error_message
: str = None,
2071 queuePosition
: int = 0,
2072 operation_state
: str = None,
2073 other_update
: dict = None,
2076 db_dict
= other_update
or {}
2077 db_dict
["queuePosition"] = queuePosition
2078 if isinstance(stage
, list):
2079 db_dict
["stage"] = stage
[0]
2080 db_dict
["detailed-status"] = " ".join(stage
)
2081 elif stage
is not None:
2082 db_dict
["stage"] = str(stage
)
2084 if error_message
is not None:
2085 db_dict
["errorMessage"] = error_message
2086 if operation_state
is not None:
2087 db_dict
["operationState"] = operation_state
2088 db_dict
["statusEnteredTime"] = time()
2089 self
.update_db_2("nslcmops", op_id
, db_dict
)
2090 except DbException
as e
:
2092 "Error writing OPERATION status for op_id: {} -> {}".format(op_id
, e
)
2095 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
2097 nsr_id
= db_nsr
["_id"]
2098 # configurationStatus
2099 config_status
= db_nsr
.get("configurationStatus")
2102 "configurationStatus.{}.status".format(index
): status
2103 for index
, v
in enumerate(config_status
)
2107 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2109 except DbException
as e
:
2111 "Error writing all configuration status, ns={}: {}".format(nsr_id
, e
)
2114 def _write_configuration_status(
2119 element_under_configuration
: str = None,
2120 element_type
: str = None,
2121 other_update
: dict = None,
2124 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
2125 # .format(vca_index, status))
2128 db_path
= "configurationStatus.{}.".format(vca_index
)
2129 db_dict
= other_update
or {}
2131 db_dict
[db_path
+ "status"] = status
2132 if element_under_configuration
:
2134 db_path
+ "elementUnderConfiguration"
2135 ] = element_under_configuration
2137 db_dict
[db_path
+ "elementType"] = element_type
2138 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2139 except DbException
as e
:
2141 "Error writing configuration status={}, ns={}, vca_index={}: {}".format(
2142 status
, nsr_id
, vca_index
, e
2146 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
2148 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
2149 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
2150 Database is used because the result can be obtained from a different LCM worker in case of HA.
2151 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
2152 :param db_nslcmop: database content of nslcmop
2153 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
2154 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
2155 computed 'vim-account-id'
2158 nslcmop_id
= db_nslcmop
["_id"]
2159 placement_engine
= deep_get(db_nslcmop
, ("operationParams", "placement-engine"))
2160 if placement_engine
== "PLA":
2162 logging_text
+ "Invoke and wait for placement optimization"
2164 await self
.msg
.aiowrite(
2165 "pla", "get_placement", {"nslcmopId": nslcmop_id
}, loop
=self
.loop
2167 db_poll_interval
= 5
2168 wait
= db_poll_interval
* 10
2170 while not pla_result
and wait
>= 0:
2171 await asyncio
.sleep(db_poll_interval
)
2172 wait
-= db_poll_interval
2173 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2174 pla_result
= deep_get(db_nslcmop
, ("_admin", "pla"))
2178 "Placement timeout for nslcmopId={}".format(nslcmop_id
)
2181 for pla_vnf
in pla_result
["vnf"]:
2182 vnfr
= db_vnfrs
.get(pla_vnf
["member-vnf-index"])
2183 if not pla_vnf
.get("vimAccountId") or not vnfr
:
2188 {"_id": vnfr
["_id"]},
2189 {"vim-account-id": pla_vnf
["vimAccountId"]},
2192 vnfr
["vim-account-id"] = pla_vnf
["vimAccountId"]
2195 def update_nsrs_with_pla_result(self
, params
):
2197 nslcmop_id
= deep_get(params
, ("placement", "nslcmopId"))
2199 "nslcmops", nslcmop_id
, {"_admin.pla": params
.get("placement")}
2201 except Exception as e
:
2202 self
.logger
.warn("Update failed for nslcmop_id={}:{}".format(nslcmop_id
, e
))
2204 async def instantiate(self
, nsr_id
, nslcmop_id
):
2207 :param nsr_id: ns instance to deploy
2208 :param nslcmop_id: operation to run
2212 # Try to lock HA task here
2213 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
2214 if not task_is_locked_by_me
:
2216 "instantiate() task is not locked by me, ns={}".format(nsr_id
)
2220 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
2221 self
.logger
.debug(logging_text
+ "Enter")
2223 # get all needed from database
2225 # database nsrs record
2228 # database nslcmops record
2231 # update operation on nsrs
2233 # update operation on nslcmops
2234 db_nslcmop_update
= {}
2236 nslcmop_operation_state
= None
2237 db_vnfrs
= {} # vnf's info indexed by member-index
2239 tasks_dict_info
= {} # from task to info text
2243 "Stage 1/5: preparation of the environment.",
2244 "Waiting for previous operations to terminate.",
2247 # ^ stage, step, VIM progress
2249 # wait for any previous tasks in process
2250 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
2252 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
2253 stage
[1] = "Reading from database."
2254 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
2255 db_nsr_update
["detailed-status"] = "creating"
2256 db_nsr_update
["operational-status"] = "init"
2257 self
._write
_ns
_status
(
2259 ns_state
="BUILDING",
2260 current_operation
="INSTANTIATING",
2261 current_operation_id
=nslcmop_id
,
2262 other_update
=db_nsr_update
,
2264 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
, queuePosition
=0)
2266 # read from db: operation
2267 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2268 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2269 ns_params
= db_nslcmop
.get("operationParams")
2270 if ns_params
and ns_params
.get("timeout_ns_deploy"):
2271 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
2273 timeout_ns_deploy
= self
.timeout
.get(
2274 "ns_deploy", self
.timeout_ns_deploy
2278 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2279 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2280 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
2281 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2282 self
.fs
.sync(db_nsr
["nsd-id"])
2284 # nsr_name = db_nsr["name"] # TODO short-name??
2286 # read from db: vnf's of this ns
2287 stage
[1] = "Getting vnfrs from db."
2288 self
.logger
.debug(logging_text
+ stage
[1])
2289 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2291 # read from db: vnfd's for every vnf
2292 db_vnfds
= [] # every vnfd data
2294 # for each vnf in ns, read vnfd
2295 for vnfr
in db_vnfrs_list
:
2296 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
2297 vnfd_id
= vnfr
["vnfd-id"]
2298 vnfd_ref
= vnfr
["vnfd-ref"]
2299 self
.fs
.sync(vnfd_id
)
2301 # if we haven't this vnfd, read it from db
2302 if vnfd_id
not in db_vnfds
:
2304 stage
[1] = "Getting vnfd={} id='{}' from db.".format(
2307 self
.logger
.debug(logging_text
+ stage
[1])
2308 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2311 db_vnfds
.append(vnfd
)
2313 # Get or generates the _admin.deployed.VCA list
2314 vca_deployed_list
= None
2315 if db_nsr
["_admin"].get("deployed"):
2316 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2317 if vca_deployed_list
is None:
2318 vca_deployed_list
= []
2319 configuration_status_list
= []
2320 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2321 db_nsr_update
["configurationStatus"] = configuration_status_list
2322 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2323 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2324 elif isinstance(vca_deployed_list
, dict):
2325 # maintain backward compatibility. Change a dict to list at database
2326 vca_deployed_list
= list(vca_deployed_list
.values())
2327 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2328 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2331 deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list
2333 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2334 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2336 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2337 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2338 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2340 "vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"}
2343 # n2vc_redesign STEP 2 Deploy Network Scenario
2344 stage
[0] = "Stage 2/5: deployment of KDUs, VMs and execution environments."
2345 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
2347 stage
[1] = "Deploying KDUs."
2348 # self.logger.debug(logging_text + "Before deploy_kdus")
2349 # Call to deploy_kdus in case exists the "vdu:kdu" param
2350 await self
.deploy_kdus(
2351 logging_text
=logging_text
,
2353 nslcmop_id
=nslcmop_id
,
2356 task_instantiation_info
=tasks_dict_info
,
2359 stage
[1] = "Getting VCA public key."
2360 # n2vc_redesign STEP 1 Get VCA public ssh-key
2361 # feature 1429. Add n2vc public key to needed VMs
2362 n2vc_key
= self
.n2vc
.get_public_key()
2363 n2vc_key_list
= [n2vc_key
]
2364 if self
.vca_config
.get("public_key"):
2365 n2vc_key_list
.append(self
.vca_config
["public_key"])
2367 stage
[1] = "Deploying NS at VIM."
2368 task_ro
= asyncio
.ensure_future(
2369 self
.instantiate_RO(
2370 logging_text
=logging_text
,
2374 db_nslcmop
=db_nslcmop
,
2377 n2vc_key_list
=n2vc_key_list
,
2381 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2382 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2384 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2385 stage
[1] = "Deploying Execution Environments."
2386 self
.logger
.debug(logging_text
+ stage
[1])
2388 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2389 for vnf_profile
in get_vnf_profiles(nsd
):
2390 vnfd_id
= vnf_profile
["vnfd-id"]
2391 vnfd
= find_in_list(db_vnfds
, lambda a_vnf
: a_vnf
["id"] == vnfd_id
)
2392 member_vnf_index
= str(vnf_profile
["id"])
2393 db_vnfr
= db_vnfrs
[member_vnf_index
]
2394 base_folder
= vnfd
["_admin"]["storage"]
2400 # Get additional parameters
2401 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
2402 if db_vnfr
.get("additionalParamsForVnf"):
2403 deploy_params
.update(
2404 parse_yaml_strings(db_vnfr
["additionalParamsForVnf"].copy())
2407 descriptor_config
= get_configuration(vnfd
, vnfd
["id"])
2408 if descriptor_config
:
2410 logging_text
=logging_text
2411 + "member_vnf_index={} ".format(member_vnf_index
),
2414 nslcmop_id
=nslcmop_id
,
2420 member_vnf_index
=member_vnf_index
,
2421 vdu_index
=vdu_index
,
2423 deploy_params
=deploy_params
,
2424 descriptor_config
=descriptor_config
,
2425 base_folder
=base_folder
,
2426 task_instantiation_info
=tasks_dict_info
,
2430 # Deploy charms for each VDU that supports one.
2431 for vdud
in get_vdu_list(vnfd
):
2433 descriptor_config
= get_configuration(vnfd
, vdu_id
)
2434 vdur
= find_in_list(
2435 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
2438 if vdur
.get("additionalParams"):
2439 deploy_params_vdu
= parse_yaml_strings(vdur
["additionalParams"])
2441 deploy_params_vdu
= deploy_params
2442 deploy_params_vdu
["OSM"] = get_osm_params(
2443 db_vnfr
, vdu_id
, vdu_count_index
=0
2445 vdud_count
= get_number_of_instances(vnfd
, vdu_id
)
2447 self
.logger
.debug("VDUD > {}".format(vdud
))
2449 "Descriptor config > {}".format(descriptor_config
)
2451 if descriptor_config
:
2454 for vdu_index
in range(vdud_count
):
2455 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2457 logging_text
=logging_text
2458 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2459 member_vnf_index
, vdu_id
, vdu_index
2463 nslcmop_id
=nslcmop_id
,
2469 member_vnf_index
=member_vnf_index
,
2470 vdu_index
=vdu_index
,
2472 deploy_params
=deploy_params_vdu
,
2473 descriptor_config
=descriptor_config
,
2474 base_folder
=base_folder
,
2475 task_instantiation_info
=tasks_dict_info
,
2478 for kdud
in get_kdu_list(vnfd
):
2479 kdu_name
= kdud
["name"]
2480 descriptor_config
= get_configuration(vnfd
, kdu_name
)
2481 if descriptor_config
:
2486 x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
2488 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
2489 if kdur
.get("additionalParams"):
2490 deploy_params_kdu
= parse_yaml_strings(
2491 kdur
["additionalParams"]
2495 logging_text
=logging_text
,
2498 nslcmop_id
=nslcmop_id
,
2504 member_vnf_index
=member_vnf_index
,
2505 vdu_index
=vdu_index
,
2507 deploy_params
=deploy_params_kdu
,
2508 descriptor_config
=descriptor_config
,
2509 base_folder
=base_folder
,
2510 task_instantiation_info
=tasks_dict_info
,
2514 # Check if this NS has a charm configuration
2515 descriptor_config
= nsd
.get("ns-configuration")
2516 if descriptor_config
and descriptor_config
.get("juju"):
2519 member_vnf_index
= None
2525 # Get additional parameters
2526 deploy_params
= {"OSM": {"vim_account_id": ns_params
["vimAccountId"]}}
2527 if db_nsr
.get("additionalParamsForNs"):
2528 deploy_params
.update(
2529 parse_yaml_strings(db_nsr
["additionalParamsForNs"].copy())
2531 base_folder
= nsd
["_admin"]["storage"]
2533 logging_text
=logging_text
,
2536 nslcmop_id
=nslcmop_id
,
2542 member_vnf_index
=member_vnf_index
,
2543 vdu_index
=vdu_index
,
2545 deploy_params
=deploy_params
,
2546 descriptor_config
=descriptor_config
,
2547 base_folder
=base_folder
,
2548 task_instantiation_info
=tasks_dict_info
,
2552 # rest of staff will be done at finally
2555 ROclient
.ROClientException
,
2561 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
)
2564 except asyncio
.CancelledError
:
2566 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
2568 exc
= "Operation was cancelled"
2569 except Exception as e
:
2570 exc
= traceback
.format_exc()
2571 self
.logger
.critical(
2572 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
2577 error_list
.append(str(exc
))
2579 # wait for pending tasks
2581 stage
[1] = "Waiting for instantiate pending tasks."
2582 self
.logger
.debug(logging_text
+ stage
[1])
2583 error_list
+= await self
._wait
_for
_tasks
(
2591 stage
[1] = stage
[2] = ""
2592 except asyncio
.CancelledError
:
2593 error_list
.append("Cancelled")
2594 # TODO cancel all tasks
2595 except Exception as exc
:
2596 error_list
.append(str(exc
))
2598 # update operation-status
2599 db_nsr_update
["operational-status"] = "running"
2600 # let's begin with VCA 'configured' status (later we can change it)
2601 db_nsr_update
["config-status"] = "configured"
2602 for task
, task_name
in tasks_dict_info
.items():
2603 if not task
.done() or task
.cancelled() or task
.exception():
2604 if task_name
.startswith(self
.task_name_deploy_vca
):
2605 # A N2VC task is pending
2606 db_nsr_update
["config-status"] = "failed"
2608 # RO or KDU task is pending
2609 db_nsr_update
["operational-status"] = "failed"
2611 # update status at database
2613 error_detail
= ". ".join(error_list
)
2614 self
.logger
.error(logging_text
+ error_detail
)
2615 error_description_nslcmop
= "{} Detail: {}".format(
2616 stage
[0], error_detail
2618 error_description_nsr
= "Operation: INSTANTIATING.{}, {}".format(
2619 nslcmop_id
, stage
[0]
2622 db_nsr_update
["detailed-status"] = (
2623 error_description_nsr
+ " Detail: " + error_detail
2625 db_nslcmop_update
["detailed-status"] = error_detail
2626 nslcmop_operation_state
= "FAILED"
2630 error_description_nsr
= error_description_nslcmop
= None
2632 db_nsr_update
["detailed-status"] = "Done"
2633 db_nslcmop_update
["detailed-status"] = "Done"
2634 nslcmop_operation_state
= "COMPLETED"
2637 self
._write
_ns
_status
(
2640 current_operation
="IDLE",
2641 current_operation_id
=None,
2642 error_description
=error_description_nsr
,
2643 error_detail
=error_detail
,
2644 other_update
=db_nsr_update
,
2646 self
._write
_op
_status
(
2649 error_message
=error_description_nslcmop
,
2650 operation_state
=nslcmop_operation_state
,
2651 other_update
=db_nslcmop_update
,
2654 if nslcmop_operation_state
:
2656 await self
.msg
.aiowrite(
2661 "nslcmop_id": nslcmop_id
,
2662 "operationState": nslcmop_operation_state
,
2666 except Exception as e
:
2668 logging_text
+ "kafka_write notification Exception {}".format(e
)
2671 self
.logger
.debug(logging_text
+ "Exit")
2672 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2674 def _get_vnfd(self
, vnfd_id
: str, cached_vnfds
: Dict
[str, Any
]):
2675 if vnfd_id
not in cached_vnfds
:
2676 cached_vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
2677 return cached_vnfds
[vnfd_id
]
2679 def _get_vnfr(self
, nsr_id
: str, vnf_profile_id
: str, cached_vnfrs
: Dict
[str, Any
]):
2680 if vnf_profile_id
not in cached_vnfrs
:
2681 cached_vnfrs
[vnf_profile_id
] = self
.db
.get_one(
2684 "member-vnf-index-ref": vnf_profile_id
,
2685 "nsr-id-ref": nsr_id
,
2688 return cached_vnfrs
[vnf_profile_id
]
2690 def _is_deployed_vca_in_relation(
2691 self
, vca
: DeployedVCA
, relation
: Relation
2694 for endpoint
in (relation
.provider
, relation
.requirer
):
2695 if endpoint
["kdu-resource-profile-id"]:
2698 vca
.vnf_profile_id
== endpoint
.vnf_profile_id
2699 and vca
.vdu_profile_id
== endpoint
.vdu_profile_id
2700 and vca
.execution_environment_ref
== endpoint
.execution_environment_ref
2706 def _update_ee_relation_data_with_implicit_data(
2707 self
, nsr_id
, nsd
, ee_relation_data
, cached_vnfds
, vnf_profile_id
: str = None
2709 ee_relation_data
= safe_get_ee_relation(
2710 nsr_id
, ee_relation_data
, vnf_profile_id
=vnf_profile_id
2712 ee_relation_level
= EELevel
.get_level(ee_relation_data
)
2713 if (ee_relation_level
in (EELevel
.VNF
, EELevel
.VDU
)) and not ee_relation_data
[
2714 "execution-environment-ref"
2716 vnf_profile
= get_vnf_profile(nsd
, ee_relation_data
["vnf-profile-id"])
2717 vnfd_id
= vnf_profile
["vnfd-id"]
2718 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2721 if ee_relation_level
== EELevel
.VNF
2722 else ee_relation_data
["vdu-profile-id"]
2724 ee
= get_juju_ee_ref(db_vnfd
, entity_id
)
2727 f
"not execution environments found for ee_relation {ee_relation_data}"
2729 ee_relation_data
["execution-environment-ref"] = ee
["id"]
2730 return ee_relation_data
2732 def _get_ns_relations(
2735 nsd
: Dict
[str, Any
],
2737 cached_vnfds
: Dict
[str, Any
],
2738 ) -> List
[Relation
]:
2740 db_ns_relations
= get_ns_configuration_relation_list(nsd
)
2741 for r
in db_ns_relations
:
2742 provider_dict
= None
2743 requirer_dict
= None
2744 if all(key
in r
for key
in ("provider", "requirer")):
2745 provider_dict
= r
["provider"]
2746 requirer_dict
= r
["requirer"]
2747 elif "entities" in r
:
2748 provider_id
= r
["entities"][0]["id"]
2751 "endpoint": r
["entities"][0]["endpoint"],
2753 if provider_id
!= nsd
["id"]:
2754 provider_dict
["vnf-profile-id"] = provider_id
2755 requirer_id
= r
["entities"][1]["id"]
2758 "endpoint": r
["entities"][1]["endpoint"],
2760 if requirer_id
!= nsd
["id"]:
2761 requirer_dict
["vnf-profile-id"] = requirer_id
2763 raise Exception("provider/requirer or entities must be included in the relation.")
2764 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2765 nsr_id
, nsd
, provider_dict
, cached_vnfds
2767 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2768 nsr_id
, nsd
, requirer_dict
, cached_vnfds
2770 provider
= EERelation(relation_provider
)
2771 requirer
= EERelation(relation_requirer
)
2772 relation
= Relation(r
["name"], provider
, requirer
)
2773 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2775 relations
.append(relation
)
2778 def _get_vnf_relations(
2781 nsd
: Dict
[str, Any
],
2783 cached_vnfds
: Dict
[str, Any
],
2784 ) -> List
[Relation
]:
2786 vnf_profile
= get_vnf_profile(nsd
, vca
.vnf_profile_id
)
2787 vnf_profile_id
= vnf_profile
["id"]
2788 vnfd_id
= vnf_profile
["vnfd-id"]
2789 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2790 db_vnf_relations
= get_relation_list(db_vnfd
, vnfd_id
)
2791 for r
in db_vnf_relations
:
2792 provider_dict
= None
2793 requirer_dict
= None
2794 if all(key
in r
for key
in ("provider", "requirer")):
2795 provider_dict
= r
["provider"]
2796 requirer_dict
= r
["requirer"]
2797 elif "entities" in r
:
2798 provider_id
= r
["entities"][0]["id"]
2801 "vnf-profile-id": vnf_profile_id
,
2802 "endpoint": r
["entities"][0]["endpoint"],
2804 if provider_id
!= vnfd_id
:
2805 provider_dict
["vdu-profile-id"] = provider_id
2806 requirer_id
= r
["entities"][1]["id"]
2809 "vnf-profile-id": vnf_profile_id
,
2810 "endpoint": r
["entities"][1]["endpoint"],
2812 if requirer_id
!= vnfd_id
:
2813 requirer_dict
["vdu-profile-id"] = requirer_id
2815 raise Exception("provider/requirer or entities must be included in the relation.")
2816 relation_provider
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2817 nsr_id
, nsd
, provider_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2819 relation_requirer
= self
._update
_ee
_relation
_data
_with
_implicit
_data
(
2820 nsr_id
, nsd
, requirer_dict
, cached_vnfds
, vnf_profile_id
=vnf_profile_id
2822 provider
= EERelation(relation_provider
)
2823 requirer
= EERelation(relation_requirer
)
2824 relation
= Relation(r
["name"], provider
, requirer
)
2825 vca_in_relation
= self
._is
_deployed
_vca
_in
_relation
(vca
, relation
)
2827 relations
.append(relation
)
2830 def _get_kdu_resource_data(
2832 ee_relation
: EERelation
,
2833 db_nsr
: Dict
[str, Any
],
2834 cached_vnfds
: Dict
[str, Any
],
2835 ) -> DeployedK8sResource
:
2836 nsd
= get_nsd(db_nsr
)
2837 vnf_profiles
= get_vnf_profiles(nsd
)
2838 vnfd_id
= find_in_list(
2840 lambda vnf_profile
: vnf_profile
["id"] == ee_relation
.vnf_profile_id
,
2842 db_vnfd
= self
._get
_vnfd
(vnfd_id
, cached_vnfds
)
2843 kdu_resource_profile
= get_kdu_resource_profile(
2844 db_vnfd
, ee_relation
.kdu_resource_profile_id
2846 kdu_name
= kdu_resource_profile
["kdu-name"]
2847 deployed_kdu
, _
= get_deployed_kdu(
2848 db_nsr
.get("_admin", ()).get("deployed", ()),
2850 ee_relation
.vnf_profile_id
,
2852 deployed_kdu
.update({"resource-name": kdu_resource_profile
["resource-name"]})
2855 def _get_deployed_component(
2857 ee_relation
: EERelation
,
2858 db_nsr
: Dict
[str, Any
],
2859 cached_vnfds
: Dict
[str, Any
],
2860 ) -> DeployedComponent
:
2861 nsr_id
= db_nsr
["_id"]
2862 deployed_component
= None
2863 ee_level
= EELevel
.get_level(ee_relation
)
2864 if ee_level
== EELevel
.NS
:
2865 vca
= get_deployed_vca(db_nsr
, {"vdu_id": None, "member-vnf-index": None})
2867 deployed_component
= DeployedVCA(nsr_id
, vca
)
2868 elif ee_level
== EELevel
.VNF
:
2869 vca
= get_deployed_vca(
2873 "member-vnf-index": ee_relation
.vnf_profile_id
,
2874 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2878 deployed_component
= DeployedVCA(nsr_id
, vca
)
2879 elif ee_level
== EELevel
.VDU
:
2880 vca
= get_deployed_vca(
2883 "vdu_id": ee_relation
.vdu_profile_id
,
2884 "member-vnf-index": ee_relation
.vnf_profile_id
,
2885 "ee_descriptor_id": ee_relation
.execution_environment_ref
,
2889 deployed_component
= DeployedVCA(nsr_id
, vca
)
2890 elif ee_level
== EELevel
.KDU
:
2891 kdu_resource_data
= self
._get
_kdu
_resource
_data
(
2892 ee_relation
, db_nsr
, cached_vnfds
2894 if kdu_resource_data
:
2895 deployed_component
= DeployedK8sResource(kdu_resource_data
)
2896 return deployed_component
2898 async def _add_relation(
2902 db_nsr
: Dict
[str, Any
],
2903 cached_vnfds
: Dict
[str, Any
],
2904 cached_vnfrs
: Dict
[str, Any
],
2906 deployed_provider
= self
._get
_deployed
_component
(
2907 relation
.provider
, db_nsr
, cached_vnfds
2909 deployed_requirer
= self
._get
_deployed
_component
(
2910 relation
.requirer
, db_nsr
, cached_vnfds
2914 and deployed_requirer
2915 and deployed_provider
.config_sw_installed
2916 and deployed_requirer
.config_sw_installed
2918 provider_db_vnfr
= (
2920 relation
.provider
.nsr_id
,
2921 relation
.provider
.vnf_profile_id
,
2924 if relation
.provider
.vnf_profile_id
2927 requirer_db_vnfr
= (
2929 relation
.requirer
.nsr_id
,
2930 relation
.requirer
.vnf_profile_id
,
2933 if relation
.requirer
.vnf_profile_id
2936 provider_vca_id
= self
.get_vca_id(provider_db_vnfr
, db_nsr
)
2937 requirer_vca_id
= self
.get_vca_id(requirer_db_vnfr
, db_nsr
)
2938 provider_relation_endpoint
= RelationEndpoint(
2939 deployed_provider
.ee_id
,
2941 relation
.provider
.endpoint
,
2943 requirer_relation_endpoint
= RelationEndpoint(
2944 deployed_requirer
.ee_id
,
2946 relation
.requirer
.endpoint
,
2948 await self
.vca_map
[vca_type
].add_relation(
2949 provider
=provider_relation_endpoint
,
2950 requirer
=requirer_relation_endpoint
,
2952 # remove entry from relations list
2956 async def _add_vca_relations(
2962 timeout
: int = 3600,
2966 # 1. find all relations for this VCA
2967 # 2. wait for other peers related
2971 # STEP 1: find all relations for this VCA
2974 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2975 nsd
= get_nsd(db_nsr
)
2978 deployed_vca_dict
= get_deployed_vca_list(db_nsr
)[vca_index
]
2979 my_vca
= DeployedVCA(nsr_id
, deployed_vca_dict
)
2984 relations
.extend(self
._get
_ns
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
2985 relations
.extend(self
._get
_vnf
_relations
(nsr_id
, nsd
, my_vca
, cached_vnfds
))
2987 # if no relations, terminate
2989 self
.logger
.debug(logging_text
+ " No relations")
2992 self
.logger
.debug(logging_text
+ " adding relations {}".format(relations
))
2999 if now
- start
>= timeout
:
3000 self
.logger
.error(logging_text
+ " : timeout adding relations")
3003 # reload nsr from database (we need to update record: _admin.deployed.VCA)
3004 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3006 # for each relation, find the VCA's related
3007 for relation
in relations
.copy():
3008 added
= await self
._add
_relation
(
3016 relations
.remove(relation
)
3019 self
.logger
.debug("Relations added")
3021 await asyncio
.sleep(5.0)
3025 except Exception as e
:
3026 self
.logger
.warn(logging_text
+ " ERROR adding relations: {}".format(e
))
3029 async def _install_kdu(
3037 k8s_instance_info
: dict,
3038 k8params
: dict = None,
3044 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
3047 "collection": "nsrs",
3048 "filter": {"_id": nsr_id
},
3049 "path": nsr_db_path
,
3052 if k8s_instance_info
.get("kdu-deployment-name"):
3053 kdu_instance
= k8s_instance_info
.get("kdu-deployment-name")
3055 kdu_instance
= self
.k8scluster_map
[
3057 ].generate_kdu_instance_name(
3058 db_dict
=db_dict_install
,
3059 kdu_model
=k8s_instance_info
["kdu-model"],
3060 kdu_name
=k8s_instance_info
["kdu-name"],
3063 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3065 await self
.k8scluster_map
[k8sclustertype
].install(
3066 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3067 kdu_model
=k8s_instance_info
["kdu-model"],
3070 db_dict
=db_dict_install
,
3072 kdu_name
=k8s_instance_info
["kdu-name"],
3073 namespace
=k8s_instance_info
["namespace"],
3074 kdu_instance
=kdu_instance
,
3078 "nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
}
3081 # Obtain services to obtain management service ip
3082 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
3083 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3084 kdu_instance
=kdu_instance
,
3085 namespace
=k8s_instance_info
["namespace"],
3088 # Obtain management service info (if exists)
3089 vnfr_update_dict
= {}
3090 kdu_config
= get_configuration(vnfd
, kdud
["name"])
3092 target_ee_list
= kdu_config
.get("execution-environment-list", [])
3097 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
3100 for service
in kdud
.get("service", [])
3101 if service
.get("mgmt-service")
3103 for mgmt_service
in mgmt_services
:
3104 for service
in services
:
3105 if service
["name"].startswith(mgmt_service
["name"]):
3106 # Mgmt service found, Obtain service ip
3107 ip
= service
.get("external_ip", service
.get("cluster_ip"))
3108 if isinstance(ip
, list) and len(ip
) == 1:
3112 "kdur.{}.ip-address".format(kdu_index
)
3115 # Check if must update also mgmt ip at the vnf
3116 service_external_cp
= mgmt_service
.get(
3117 "external-connection-point-ref"
3119 if service_external_cp
:
3121 deep_get(vnfd
, ("mgmt-interface", "cp"))
3122 == service_external_cp
3124 vnfr_update_dict
["ip-address"] = ip
3129 "external-connection-point-ref", ""
3131 == service_external_cp
,
3134 "kdur.{}.ip-address".format(kdu_index
)
3139 "Mgmt service name: {} not found".format(
3140 mgmt_service
["name"]
3144 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
3145 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
3147 kdu_config
= get_configuration(vnfd
, k8s_instance_info
["kdu-name"])
3150 and kdu_config
.get("initial-config-primitive")
3151 and get_juju_ee_ref(vnfd
, k8s_instance_info
["kdu-name"]) is None
3153 initial_config_primitive_list
= kdu_config
.get(
3154 "initial-config-primitive"
3156 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
3158 for initial_config_primitive
in initial_config_primitive_list
:
3159 primitive_params_
= self
._map
_primitive
_params
(
3160 initial_config_primitive
, {}, {}
3163 await asyncio
.wait_for(
3164 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
3165 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
3166 kdu_instance
=kdu_instance
,
3167 primitive_name
=initial_config_primitive
["name"],
3168 params
=primitive_params_
,
3169 db_dict
=db_dict_install
,
3175 except Exception as e
:
3176 # Prepare update db with error and raise exception
3179 "nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)}
3183 vnfr_data
.get("_id"),
3184 {"kdur.{}.status".format(kdu_index
): "ERROR"},
3187 # ignore to keep original exception
3189 # reraise original error
3194 async def deploy_kdus(
3201 task_instantiation_info
,
3203 # Launch kdus if present in the descriptor
3205 k8scluster_id_2_uuic
= {
3206 "helm-chart-v3": {},
3211 async def _get_cluster_id(cluster_id
, cluster_type
):
3212 nonlocal k8scluster_id_2_uuic
3213 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
3214 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
3216 # check if K8scluster is creating and wait look if previous tasks in process
3217 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related(
3218 "k8scluster", cluster_id
3221 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(
3222 task_name
, cluster_id
3224 self
.logger
.debug(logging_text
+ text
)
3225 await asyncio
.wait(task_dependency
, timeout
=3600)
3227 db_k8scluster
= self
.db
.get_one(
3228 "k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False
3230 if not db_k8scluster
:
3231 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
3233 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
3235 if cluster_type
== "helm-chart-v3":
3237 # backward compatibility for existing clusters that have not been initialized for helm v3
3238 k8s_credentials
= yaml
.safe_dump(
3239 db_k8scluster
.get("credentials")
3241 k8s_id
, uninstall_sw
= await self
.k8sclusterhelm3
.init_env(
3242 k8s_credentials
, reuse_cluster_uuid
=cluster_id
3244 db_k8scluster_update
= {}
3245 db_k8scluster_update
["_admin.helm-chart-v3.error_msg"] = None
3246 db_k8scluster_update
["_admin.helm-chart-v3.id"] = k8s_id
3247 db_k8scluster_update
[
3248 "_admin.helm-chart-v3.created"
3250 db_k8scluster_update
[
3251 "_admin.helm-chart-v3.operationalState"
3254 "k8sclusters", cluster_id
, db_k8scluster_update
3256 except Exception as e
:
3259 + "error initializing helm-v3 cluster: {}".format(str(e
))
3262 "K8s cluster '{}' has not been initialized for '{}'".format(
3263 cluster_id
, cluster_type
3268 "K8s cluster '{}' has not been initialized for '{}'".format(
3269 cluster_id
, cluster_type
3272 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
3275 logging_text
+= "Deploy kdus: "
3278 db_nsr_update
= {"_admin.deployed.K8s": []}
3279 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3282 updated_cluster_list
= []
3283 updated_v3_cluster_list
= []
3285 for vnfr_data
in db_vnfrs
.values():
3286 vca_id
= self
.get_vca_id(vnfr_data
, {})
3287 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
3288 # Step 0: Prepare and set parameters
3289 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
3290 vnfd_id
= vnfr_data
.get("vnfd-id")
3291 vnfd_with_id
= find_in_list(
3292 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3296 for kdud
in vnfd_with_id
["kdu"]
3297 if kdud
["name"] == kdur
["kdu-name"]
3299 namespace
= kdur
.get("k8s-namespace")
3300 kdu_deployment_name
= kdur
.get("kdu-deployment-name")
3301 if kdur
.get("helm-chart"):
3302 kdumodel
= kdur
["helm-chart"]
3303 # Default version: helm3, if helm-version is v2 assign v2
3304 k8sclustertype
= "helm-chart-v3"
3305 self
.logger
.debug("kdur: {}".format(kdur
))
3307 kdur
.get("helm-version")
3308 and kdur
.get("helm-version") == "v2"
3310 k8sclustertype
= "helm-chart"
3311 elif kdur
.get("juju-bundle"):
3312 kdumodel
= kdur
["juju-bundle"]
3313 k8sclustertype
= "juju-bundle"
3316 "kdu type for kdu='{}.{}' is neither helm-chart nor "
3317 "juju-bundle. Maybe an old NBI version is running".format(
3318 vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]
3321 # check if kdumodel is a file and exists
3323 vnfd_with_id
= find_in_list(
3324 db_vnfds
, lambda vnfd
: vnfd
["_id"] == vnfd_id
3326 storage
= deep_get(vnfd_with_id
, ("_admin", "storage"))
3327 if storage
: # may be not present if vnfd has not artifacts
3328 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
3329 if storage
["pkg-dir"]:
3330 filename
= "{}/{}/{}s/{}".format(
3337 filename
= "{}/Scripts/{}s/{}".format(
3342 if self
.fs
.file_exists(
3343 filename
, mode
="file"
3344 ) or self
.fs
.file_exists(filename
, mode
="dir"):
3345 kdumodel
= self
.fs
.path
+ filename
3346 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
3348 except Exception: # it is not a file
3351 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
3352 step
= "Synchronize repos for k8s cluster '{}'".format(
3355 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
3359 k8sclustertype
== "helm-chart"
3360 and cluster_uuid
not in updated_cluster_list
3362 k8sclustertype
== "helm-chart-v3"
3363 and cluster_uuid
not in updated_v3_cluster_list
3365 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
3366 self
.k8scluster_map
[k8sclustertype
].synchronize_repos(
3367 cluster_uuid
=cluster_uuid
3370 if del_repo_list
or added_repo_dict
:
3371 if k8sclustertype
== "helm-chart":
3373 "_admin.helm_charts_added." + item
: None
3374 for item
in del_repo_list
3377 "_admin.helm_charts_added." + item
: name
3378 for item
, name
in added_repo_dict
.items()
3380 updated_cluster_list
.append(cluster_uuid
)
3381 elif k8sclustertype
== "helm-chart-v3":
3383 "_admin.helm_charts_v3_added." + item
: None
3384 for item
in del_repo_list
3387 "_admin.helm_charts_v3_added." + item
: name
3388 for item
, name
in added_repo_dict
.items()
3390 updated_v3_cluster_list
.append(cluster_uuid
)
3392 logging_text
+ "repos synchronized on k8s cluster "
3393 "'{}' to_delete: {}, to_add: {}".format(
3394 k8s_cluster_id
, del_repo_list
, added_repo_dict
3399 {"_id": k8s_cluster_id
},
3405 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(
3406 vnfr_data
["member-vnf-index-ref"],
3410 k8s_instance_info
= {
3411 "kdu-instance": None,
3412 "k8scluster-uuid": cluster_uuid
,
3413 "k8scluster-type": k8sclustertype
,
3414 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
3415 "kdu-name": kdur
["kdu-name"],
3416 "kdu-model": kdumodel
,
3417 "namespace": namespace
,
3418 "kdu-deployment-name": kdu_deployment_name
,
3420 db_path
= "_admin.deployed.K8s.{}".format(index
)
3421 db_nsr_update
[db_path
] = k8s_instance_info
3422 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3423 vnfd_with_id
= find_in_list(
3424 db_vnfds
, lambda vnf
: vnf
["_id"] == vnfd_id
3426 task
= asyncio
.ensure_future(
3435 k8params
=desc_params
,
3440 self
.lcm_tasks
.register(
3444 "instantiate_KDU-{}".format(index
),
3447 task_instantiation_info
[task
] = "Deploying KDU {}".format(
3453 except (LcmException
, asyncio
.CancelledError
):
3455 except Exception as e
:
3456 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
3457 if isinstance(e
, (N2VCException
, DbException
)):
3458 self
.logger
.error(logging_text
+ msg
)
3460 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
3461 raise LcmException(msg
)
3464 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3483 task_instantiation_info
,
3486 # launch instantiate_N2VC in a asyncio task and register task object
3487 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
3488 # if not found, create one entry and update database
3489 # fill db_nsr._admin.deployed.VCA.<index>
3492 logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
)
3494 if "execution-environment-list" in descriptor_config
:
3495 ee_list
= descriptor_config
.get("execution-environment-list", [])
3496 elif "juju" in descriptor_config
:
3497 ee_list
= [descriptor_config
] # ns charms
3498 else: # other types as script are not supported
3501 for ee_item
in ee_list
:
3504 + "_deploy_n2vc ee_item juju={}, helm={}".format(
3505 ee_item
.get("juju"), ee_item
.get("helm-chart")
3508 ee_descriptor_id
= ee_item
.get("id")
3509 if ee_item
.get("juju"):
3510 vca_name
= ee_item
["juju"].get("charm")
3513 if ee_item
["juju"].get("charm") is not None
3516 if ee_item
["juju"].get("cloud") == "k8s":
3517 vca_type
= "k8s_proxy_charm"
3518 elif ee_item
["juju"].get("proxy") is False:
3519 vca_type
= "native_charm"
3520 elif ee_item
.get("helm-chart"):
3521 vca_name
= ee_item
["helm-chart"]
3522 if ee_item
.get("helm-version") and ee_item
.get("helm-version") == "v2":
3525 vca_type
= "helm-v3"
3528 logging_text
+ "skipping non juju neither charm configuration"
3533 for vca_index
, vca_deployed
in enumerate(
3534 db_nsr
["_admin"]["deployed"]["VCA"]
3536 if not vca_deployed
:
3539 vca_deployed
.get("member-vnf-index") == member_vnf_index
3540 and vca_deployed
.get("vdu_id") == vdu_id
3541 and vca_deployed
.get("kdu_name") == kdu_name
3542 and vca_deployed
.get("vdu_count_index", 0) == vdu_index
3543 and vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
3547 # not found, create one.
3549 "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
3552 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
3554 target
+= "/kdu/{}".format(kdu_name
)
3556 "target_element": target
,
3557 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
3558 "member-vnf-index": member_vnf_index
,
3560 "kdu_name": kdu_name
,
3561 "vdu_count_index": vdu_index
,
3562 "operational-status": "init", # TODO revise
3563 "detailed-status": "", # TODO revise
3564 "step": "initial-deploy", # TODO revise
3566 "vdu_name": vdu_name
,
3568 "ee_descriptor_id": ee_descriptor_id
,
3572 # create VCA and configurationStatus in db
3574 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
3575 "configurationStatus.{}".format(vca_index
): dict(),
3577 self
.update_db_2("nsrs", nsr_id
, db_dict
)
3579 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
3581 self
.logger
.debug("N2VC > NSR_ID > {}".format(nsr_id
))
3582 self
.logger
.debug("N2VC > DB_NSR > {}".format(db_nsr
))
3583 self
.logger
.debug("N2VC > VCA_DEPLOYED > {}".format(vca_deployed
))
3586 task_n2vc
= asyncio
.ensure_future(
3587 self
.instantiate_N2VC(
3588 logging_text
=logging_text
,
3589 vca_index
=vca_index
,
3595 vdu_index
=vdu_index
,
3596 deploy_params
=deploy_params
,
3597 config_descriptor
=descriptor_config
,
3598 base_folder
=base_folder
,
3599 nslcmop_id
=nslcmop_id
,
3603 ee_config_descriptor
=ee_item
,
3606 self
.lcm_tasks
.register(
3610 "instantiate_N2VC-{}".format(vca_index
),
3613 task_instantiation_info
[
3615 ] = self
.task_name_deploy_vca
+ " {}.{}".format(
3616 member_vnf_index
or "", vdu_id
or ""
3620 def _create_nslcmop(nsr_id
, operation
, params
):
3622 Creates a ns-lcm-opp content to be stored at database.
3623 :param nsr_id: internal id of the instance
3624 :param operation: instantiate, terminate, scale, action, ...
3625 :param params: user parameters for the operation
3626 :return: dictionary following SOL005 format
3628 # Raise exception if invalid arguments
3629 if not (nsr_id
and operation
and params
):
3631 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided"
3638 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
3639 "operationState": "PROCESSING",
3640 "statusEnteredTime": now
,
3641 "nsInstanceId": nsr_id
,
3642 "lcmOperationType": operation
,
3644 "isAutomaticInvocation": False,
3645 "operationParams": params
,
3646 "isCancelPending": False,
3648 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
3649 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
3654 def _format_additional_params(self
, params
):
3655 params
= params
or {}
3656 for key
, value
in params
.items():
3657 if str(value
).startswith("!!yaml "):
3658 params
[key
] = yaml
.safe_load(value
[7:])
3661 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
3662 primitive
= seq
.get("name")
3663 primitive_params
= {}
3665 "member_vnf_index": vnf_index
,
3666 "primitive": primitive
,
3667 "primitive_params": primitive_params
,
3670 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
3674 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
3675 op
= deep_get(db_nslcmop
, ("_admin", "operations"), [])[op_index
]
3676 if op
.get("operationState") == "COMPLETED":
3677 # b. Skip sub-operation
3678 # _ns_execute_primitive() or RO.create_action() will NOT be executed
3679 return self
.SUBOPERATION_STATUS_SKIP
3681 # c. retry executing sub-operation
3682 # The sub-operation exists, and operationState != 'COMPLETED'
3683 # Update operationState = 'PROCESSING' to indicate a retry.
3684 operationState
= "PROCESSING"
3685 detailed_status
= "In progress"
3686 self
._update
_suboperation
_status
(
3687 db_nslcmop
, op_index
, operationState
, detailed_status
3689 # Return the sub-operation index
3690 # _ns_execute_primitive() or RO.create_action() will be called from scale()
3691 # with arguments extracted from the sub-operation
3694 # Find a sub-operation where all keys in a matching dictionary must match
3695 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
3696 def _find_suboperation(self
, db_nslcmop
, match
):
3697 if db_nslcmop
and match
:
3698 op_list
= db_nslcmop
.get("_admin", {}).get("operations", [])
3699 for i
, op
in enumerate(op_list
):
3700 if all(op
.get(k
) == match
[k
] for k
in match
):
3702 return self
.SUBOPERATION_STATUS_NOT_FOUND
3704 # Update status for a sub-operation given its index
3705 def _update_suboperation_status(
3706 self
, db_nslcmop
, op_index
, operationState
, detailed_status
3708 # Update DB for HA tasks
3709 q_filter
= {"_id": db_nslcmop
["_id"]}
3711 "_admin.operations.{}.operationState".format(op_index
): operationState
,
3712 "_admin.operations.{}.detailed-status".format(op_index
): detailed_status
,
3715 "nslcmops", q_filter
=q_filter
, update_dict
=update_dict
, fail_on_empty
=False
3718 # Add sub-operation, return the index of the added sub-operation
3719 # Optionally, set operationState, detailed-status, and operationType
3720 # Status and type are currently set for 'scale' sub-operations:
3721 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
3722 # 'detailed-status' : status message
3723 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
3724 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
3725 def _add_suboperation(
3733 mapped_primitive_params
,
3734 operationState
=None,
3735 detailed_status
=None,
3738 RO_scaling_info
=None,
3741 return self
.SUBOPERATION_STATUS_NOT_FOUND
3742 # Get the "_admin.operations" list, if it exists
3743 db_nslcmop_admin
= db_nslcmop
.get("_admin", {})
3744 op_list
= db_nslcmop_admin
.get("operations")
3745 # Create or append to the "_admin.operations" list
3747 "member_vnf_index": vnf_index
,
3749 "vdu_count_index": vdu_count_index
,
3750 "primitive": primitive
,
3751 "primitive_params": mapped_primitive_params
,
3754 new_op
["operationState"] = operationState
3756 new_op
["detailed-status"] = detailed_status
3758 new_op
["lcmOperationType"] = operationType
3760 new_op
["RO_nsr_id"] = RO_nsr_id
3762 new_op
["RO_scaling_info"] = RO_scaling_info
3764 # No existing operations, create key 'operations' with current operation as first list element
3765 db_nslcmop_admin
.update({"operations": [new_op
]})
3766 op_list
= db_nslcmop_admin
.get("operations")
3768 # Existing operations, append operation to list
3769 op_list
.append(new_op
)
3771 db_nslcmop_update
= {"_admin.operations": op_list
}
3772 self
.update_db_2("nslcmops", db_nslcmop
["_id"], db_nslcmop_update
)
3773 op_index
= len(op_list
) - 1
3776 # Helper methods for scale() sub-operations
3778 # pre-scale/post-scale:
3779 # Check for 3 different cases:
3780 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
3781 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
3782 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
3783 def _check_or_add_scale_suboperation(
3787 vnf_config_primitive
,
3791 RO_scaling_info
=None,
3793 # Find this sub-operation
3794 if RO_nsr_id
and RO_scaling_info
:
3795 operationType
= "SCALE-RO"
3797 "member_vnf_index": vnf_index
,
3798 "RO_nsr_id": RO_nsr_id
,
3799 "RO_scaling_info": RO_scaling_info
,
3803 "member_vnf_index": vnf_index
,
3804 "primitive": vnf_config_primitive
,
3805 "primitive_params": primitive_params
,
3806 "lcmOperationType": operationType
,
3808 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
3809 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
3810 # a. New sub-operation
3811 # The sub-operation does not exist, add it.
3812 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
3813 # The following parameters are set to None for all kind of scaling:
3815 vdu_count_index
= None
3817 if RO_nsr_id
and RO_scaling_info
:
3818 vnf_config_primitive
= None
3819 primitive_params
= None
3822 RO_scaling_info
= None
3823 # Initial status for sub-operation
3824 operationState
= "PROCESSING"
3825 detailed_status
= "In progress"
3826 # Add sub-operation for pre/post-scaling (zero or more operations)
3827 self
._add
_suboperation
(
3833 vnf_config_primitive
,
3841 return self
.SUBOPERATION_STATUS_NEW
3843 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3844 # or op_index (operationState != 'COMPLETED')
3845 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3847 # Function to return execution_environment id
3849 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3850 # TODO vdu_index_count
3851 for vca
in vca_deployed_list
:
3852 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3855 async def destroy_N2VC(
3863 exec_primitives
=True,
3868 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3869 :param logging_text:
3871 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3872 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3873 :param vca_index: index in the database _admin.deployed.VCA
3874 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3875 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3876 not executed properly
3877 :param scaling_in: True destroys the application, False destroys the model
3878 :return: None or exception
3883 + " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3884 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3888 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3890 # execute terminate_primitives
3892 terminate_primitives
= get_ee_sorted_terminate_config_primitive_list(
3893 config_descriptor
.get("terminate-config-primitive"),
3894 vca_deployed
.get("ee_descriptor_id"),
3896 vdu_id
= vca_deployed
.get("vdu_id")
3897 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3898 vdu_name
= vca_deployed
.get("vdu_name")
3899 vnf_index
= vca_deployed
.get("member-vnf-index")
3900 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3901 for seq
in terminate_primitives
:
3902 # For each sequence in list, get primitive and call _ns_execute_primitive()
3903 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3904 vnf_index
, seq
.get("name")
3906 self
.logger
.debug(logging_text
+ step
)
3907 # Create the primitive for each sequence, i.e. "primitive": "touch"
3908 primitive
= seq
.get("name")
3909 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(
3914 self
._add
_suboperation
(
3921 mapped_primitive_params
,
3923 # Sub-operations: Call _ns_execute_primitive() instead of action()
3925 result
, result_detail
= await self
._ns
_execute
_primitive
(
3926 vca_deployed
["ee_id"],
3928 mapped_primitive_params
,
3932 except LcmException
:
3933 # this happens when VCA is not deployed. In this case it is not needed to terminate
3935 result_ok
= ["COMPLETED", "PARTIALLY_COMPLETED"]
3936 if result
not in result_ok
:
3938 "terminate_primitive {} for vnf_member_index={} fails with "
3939 "error {}".format(seq
.get("name"), vnf_index
, result_detail
)
3941 # set that this VCA do not need terminated
3942 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(
3946 "nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False}
3949 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3950 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3953 await self
.vca_map
[vca_type
].delete_execution_environment(
3954 vca_deployed
["ee_id"],
3955 scaling_in
=scaling_in
,
3960 async def _delete_all_N2VC(self
, db_nsr
: dict, vca_id
: str = None):
3961 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="TERMINATING")
3962 namespace
= "." + db_nsr
["_id"]
3964 await self
.n2vc
.delete_namespace(
3965 namespace
=namespace
,
3966 total_timeout
=self
.timeout_charm_delete
,
3969 except N2VCNotFound
: # already deleted. Skip
3971 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
="DELETED")
3973 async def _terminate_RO(
3974 self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
3977 Terminates a deployment from RO
3978 :param logging_text:
3979 :param nsr_deployed: db_nsr._admin.deployed
3982 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3983 this method will update only the index 2, but it will write on database the concatenated content of the list
3988 ro_nsr_id
= ro_delete_action
= None
3989 if nsr_deployed
and nsr_deployed
.get("RO"):
3990 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3991 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3994 stage
[2] = "Deleting ns from VIM."
3995 db_nsr_update
["detailed-status"] = " ".join(stage
)
3996 self
._write
_op
_status
(nslcmop_id
, stage
)
3997 self
.logger
.debug(logging_text
+ stage
[2])
3998 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3999 self
._write
_op
_status
(nslcmop_id
, stage
)
4000 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
4001 ro_delete_action
= desc
["action_id"]
4003 "_admin.deployed.RO.nsr_delete_action_id"
4004 ] = ro_delete_action
4005 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4006 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4007 if ro_delete_action
:
4008 # wait until NS is deleted from VIM
4009 stage
[2] = "Waiting ns deleted from VIM."
4010 detailed_status_old
= None
4014 + " RO_id={} ro_delete_action={}".format(
4015 ro_nsr_id
, ro_delete_action
4018 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4019 self
._write
_op
_status
(nslcmop_id
, stage
)
4021 delete_timeout
= 20 * 60 # 20 minutes
4022 while delete_timeout
> 0:
4023 desc
= await self
.RO
.show(
4025 item_id_name
=ro_nsr_id
,
4026 extra_item
="action",
4027 extra_item_id
=ro_delete_action
,
4031 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4033 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4034 if ns_status
== "ERROR":
4035 raise ROclient
.ROClientException(ns_status_info
)
4036 elif ns_status
== "BUILD":
4037 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
4038 elif ns_status
== "ACTIVE":
4039 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4040 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4045 ), "ROclient.check_action_status returns unknown {}".format(
4048 if stage
[2] != detailed_status_old
:
4049 detailed_status_old
= stage
[2]
4050 db_nsr_update
["detailed-status"] = " ".join(stage
)
4051 self
._write
_op
_status
(nslcmop_id
, stage
)
4052 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4053 await asyncio
.sleep(5, loop
=self
.loop
)
4055 else: # delete_timeout <= 0:
4056 raise ROclient
.ROClientException(
4057 "Timeout waiting ns deleted from VIM"
4060 except Exception as e
:
4061 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4063 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4065 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
4066 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
4067 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
4069 logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
)
4072 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4074 failed_detail
.append("delete conflict: {}".format(e
))
4077 + "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
)
4080 failed_detail
.append("delete error: {}".format(e
))
4082 logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
)
4086 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
4087 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
4089 stage
[2] = "Deleting nsd from RO."
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
)
4093 await self
.RO
.delete("nsd", ro_nsd_id
)
4095 logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
)
4097 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4098 except Exception as e
:
4100 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4102 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
4104 logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
)
4107 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4109 failed_detail
.append(
4110 "ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
)
4112 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4114 failed_detail
.append(
4115 "ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
)
4117 self
.logger
.error(logging_text
+ failed_detail
[-1])
4119 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
4120 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
4121 if not vnf_deployed
or not vnf_deployed
["id"]:
4124 ro_vnfd_id
= vnf_deployed
["id"]
4127 ] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
4128 vnf_deployed
["member-vnf-index"], ro_vnfd_id
4130 db_nsr_update
["detailed-status"] = " ".join(stage
)
4131 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4132 self
._write
_op
_status
(nslcmop_id
, stage
)
4133 await self
.RO
.delete("vnfd", ro_vnfd_id
)
4135 logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
)
4137 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
4138 except Exception as e
:
4140 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404
4143 "_admin.deployed.RO.vnfd.{}.id".format(index
)
4147 + "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
)
4150 isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409
4152 failed_detail
.append(
4153 "ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
)
4155 self
.logger
.debug(logging_text
+ failed_detail
[-1])
4157 failed_detail
.append(
4158 "ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
)
4160 self
.logger
.error(logging_text
+ failed_detail
[-1])
4163 stage
[2] = "Error deleting from VIM"
4165 stage
[2] = "Deleted from VIM"
4166 db_nsr_update
["detailed-status"] = " ".join(stage
)
4167 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4168 self
._write
_op
_status
(nslcmop_id
, stage
)
4171 raise LcmException("; ".join(failed_detail
))
4173 async def terminate(self
, nsr_id
, nslcmop_id
):
4174 # Try to lock HA task here
4175 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4176 if not task_is_locked_by_me
:
4179 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
4180 self
.logger
.debug(logging_text
+ "Enter")
4181 timeout_ns_terminate
= self
.timeout_ns_terminate
4184 operation_params
= None
4186 error_list
= [] # annotates all failed error messages
4187 db_nslcmop_update
= {}
4188 autoremove
= False # autoremove after terminated
4189 tasks_dict_info
= {}
4192 "Stage 1/3: Preparing task.",
4193 "Waiting for previous operations to terminate.",
4196 # ^ contains [stage, step, VIM-status]
4198 # wait for any previous tasks in process
4199 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4201 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
4202 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4203 operation_params
= db_nslcmop
.get("operationParams") or {}
4204 if operation_params
.get("timeout_ns_terminate"):
4205 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
4206 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
4207 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4209 db_nsr_update
["operational-status"] = "terminating"
4210 db_nsr_update
["config-status"] = "terminating"
4211 self
._write
_ns
_status
(
4213 ns_state
="TERMINATING",
4214 current_operation
="TERMINATING",
4215 current_operation_id
=nslcmop_id
,
4216 other_update
=db_nsr_update
,
4218 self
._write
_op
_status
(op_id
=nslcmop_id
, queuePosition
=0, stage
=stage
)
4219 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
4220 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
4223 stage
[1] = "Getting vnf descriptors from db."
4224 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
4226 db_vnfr
["member-vnf-index-ref"]: db_vnfr
for db_vnfr
in db_vnfrs_list
4228 db_vnfds_from_id
= {}
4229 db_vnfds_from_member_index
= {}
4231 for vnfr
in db_vnfrs_list
:
4232 vnfd_id
= vnfr
["vnfd-id"]
4233 if vnfd_id
not in db_vnfds_from_id
:
4234 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
4235 db_vnfds_from_id
[vnfd_id
] = vnfd
4236 db_vnfds_from_member_index
[
4237 vnfr
["member-vnf-index-ref"]
4238 ] = db_vnfds_from_id
[vnfd_id
]
4240 # Destroy individual execution environments when there are terminating primitives.
4241 # Rest of EE will be deleted at once
4242 # TODO - check before calling _destroy_N2VC
4243 # if not operation_params.get("skip_terminate_primitives"):#
4244 # or not vca.get("needed_terminate"):
4245 stage
[0] = "Stage 2/3 execute terminating primitives."
4246 self
.logger
.debug(logging_text
+ stage
[0])
4247 stage
[1] = "Looking execution environment that needs terminate."
4248 self
.logger
.debug(logging_text
+ stage
[1])
4250 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
4251 config_descriptor
= None
4252 vca_member_vnf_index
= vca
.get("member-vnf-index")
4253 vca_id
= self
.get_vca_id(
4254 db_vnfrs_dict
.get(vca_member_vnf_index
)
4255 if vca_member_vnf_index
4259 if not vca
or not vca
.get("ee_id"):
4261 if not vca
.get("member-vnf-index"):
4263 config_descriptor
= db_nsr
.get("ns-configuration")
4264 elif vca
.get("vdu_id"):
4265 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4266 config_descriptor
= get_configuration(db_vnfd
, vca
.get("vdu_id"))
4267 elif vca
.get("kdu_name"):
4268 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4269 config_descriptor
= get_configuration(db_vnfd
, vca
.get("kdu_name"))
4271 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
4272 config_descriptor
= get_configuration(db_vnfd
, db_vnfd
["id"])
4273 vca_type
= vca
.get("type")
4274 exec_terminate_primitives
= not operation_params
.get(
4275 "skip_terminate_primitives"
4276 ) and vca
.get("needed_terminate")
4277 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
4278 # pending native charms
4280 True if vca_type
in ("helm", "helm-v3", "native_charm") else False
4282 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
4283 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
4284 task
= asyncio
.ensure_future(
4292 exec_terminate_primitives
,
4296 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
4298 # wait for pending tasks of terminate primitives
4302 + "Waiting for tasks {}".format(list(tasks_dict_info
.keys()))
4304 error_list
= await self
._wait
_for
_tasks
(
4307 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
4311 tasks_dict_info
.clear()
4313 return # raise LcmException("; ".join(error_list))
4315 # remove All execution environments at once
4316 stage
[0] = "Stage 3/3 delete all."
4318 if nsr_deployed
.get("VCA"):
4319 stage
[1] = "Deleting all execution environments."
4320 self
.logger
.debug(logging_text
+ stage
[1])
4321 vca_id
= self
.get_vca_id({}, db_nsr
)
4322 task_delete_ee
= asyncio
.ensure_future(
4324 self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
, vca_id
=vca_id
),
4325 timeout
=self
.timeout_charm_delete
,
4328 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
4329 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
4331 # Delete from k8scluster
4332 stage
[1] = "Deleting KDUs."
4333 self
.logger
.debug(logging_text
+ stage
[1])
4334 # print(nsr_deployed)
4335 for kdu
in get_iterable(nsr_deployed
, "K8s"):
4336 if not kdu
or not kdu
.get("kdu-instance"):
4338 kdu_instance
= kdu
.get("kdu-instance")
4339 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
4340 # TODO: Uninstall kdu instances taking into account they could be deployed in different VIMs
4341 vca_id
= self
.get_vca_id({}, db_nsr
)
4342 task_delete_kdu_instance
= asyncio
.ensure_future(
4343 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
4344 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4345 kdu_instance
=kdu_instance
,
4352 + "Unknown k8s deployment type {}".format(
4353 kdu
.get("k8scluster-type")
4358 task_delete_kdu_instance
4359 ] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
4362 stage
[1] = "Deleting ns from VIM."
4364 task_delete_ro
= asyncio
.ensure_future(
4365 self
._terminate
_ng
_ro
(
4366 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4370 task_delete_ro
= asyncio
.ensure_future(
4372 logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
4375 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
4377 # rest of staff will be done at finally
4380 ROclient
.ROClientException
,
4385 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4387 except asyncio
.CancelledError
:
4389 logging_text
+ "Cancelled Exception while '{}'".format(stage
[1])
4391 exc
= "Operation was cancelled"
4392 except Exception as e
:
4393 exc
= traceback
.format_exc()
4394 self
.logger
.critical(
4395 logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
),
4400 error_list
.append(str(exc
))
4402 # wait for pending tasks
4404 stage
[1] = "Waiting for terminate pending tasks."
4405 self
.logger
.debug(logging_text
+ stage
[1])
4406 error_list
+= await self
._wait
_for
_tasks
(
4409 timeout_ns_terminate
,
4413 stage
[1] = stage
[2] = ""
4414 except asyncio
.CancelledError
:
4415 error_list
.append("Cancelled")
4416 # TODO cancell all tasks
4417 except Exception as exc
:
4418 error_list
.append(str(exc
))
4419 # update status at database
4421 error_detail
= "; ".join(error_list
)
4422 # self.logger.error(logging_text + error_detail)
4423 error_description_nslcmop
= "{} Detail: {}".format(
4424 stage
[0], error_detail
4426 error_description_nsr
= "Operation: TERMINATING.{}, {}.".format(
4427 nslcmop_id
, stage
[0]
4430 db_nsr_update
["operational-status"] = "failed"
4431 db_nsr_update
["detailed-status"] = (
4432 error_description_nsr
+ " Detail: " + error_detail
4434 db_nslcmop_update
["detailed-status"] = error_detail
4435 nslcmop_operation_state
= "FAILED"
4439 error_description_nsr
= error_description_nslcmop
= None
4440 ns_state
= "NOT_INSTANTIATED"
4441 db_nsr_update
["operational-status"] = "terminated"
4442 db_nsr_update
["detailed-status"] = "Done"
4443 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
4444 db_nslcmop_update
["detailed-status"] = "Done"
4445 nslcmop_operation_state
= "COMPLETED"
4448 self
._write
_ns
_status
(
4451 current_operation
="IDLE",
4452 current_operation_id
=None,
4453 error_description
=error_description_nsr
,
4454 error_detail
=error_detail
,
4455 other_update
=db_nsr_update
,
4457 self
._write
_op
_status
(
4460 error_message
=error_description_nslcmop
,
4461 operation_state
=nslcmop_operation_state
,
4462 other_update
=db_nslcmop_update
,
4464 if ns_state
== "NOT_INSTANTIATED":
4468 {"nsr-id-ref": nsr_id
},
4469 {"_admin.nsState": "NOT_INSTANTIATED"},
4471 except DbException
as e
:
4474 + "Error writing VNFR status for nsr-id-ref: {} -> {}".format(
4478 if operation_params
:
4479 autoremove
= operation_params
.get("autoremove", False)
4480 if nslcmop_operation_state
:
4482 await self
.msg
.aiowrite(
4487 "nslcmop_id": nslcmop_id
,
4488 "operationState": nslcmop_operation_state
,
4489 "autoremove": autoremove
,
4493 except Exception as e
:
4495 logging_text
+ "kafka_write notification Exception {}".format(e
)
4498 self
.logger
.debug(logging_text
+ "Exit")
4499 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
4501 async def _wait_for_tasks(
4502 self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None
4505 error_detail_list
= []
4507 pending_tasks
= list(created_tasks_info
.keys())
4508 num_tasks
= len(pending_tasks
)
4510 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4511 self
._write
_op
_status
(nslcmop_id
, stage
)
4512 while pending_tasks
:
4514 _timeout
= timeout
+ time_start
- time()
4515 done
, pending_tasks
= await asyncio
.wait(
4516 pending_tasks
, timeout
=_timeout
, return_when
=asyncio
.FIRST_COMPLETED
4518 num_done
+= len(done
)
4519 if not done
: # Timeout
4520 for task
in pending_tasks
:
4521 new_error
= created_tasks_info
[task
] + ": Timeout"
4522 error_detail_list
.append(new_error
)
4523 error_list
.append(new_error
)
4526 if task
.cancelled():
4529 exc
= task
.exception()
4531 if isinstance(exc
, asyncio
.TimeoutError
):
4533 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
4534 error_list
.append(created_tasks_info
[task
])
4535 error_detail_list
.append(new_error
)
4542 ROclient
.ROClientException
,
4548 self
.logger
.error(logging_text
+ new_error
)
4550 exc_traceback
= "".join(
4551 traceback
.format_exception(None, exc
, exc
.__traceback
__)
4555 + created_tasks_info
[task
]
4561 logging_text
+ created_tasks_info
[task
] + ": Done"
4563 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
4565 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
4566 if nsr_id
: # update also nsr
4571 "errorDescription": "Error at: " + ", ".join(error_list
),
4572 "errorDetail": ". ".join(error_detail_list
),
4575 self
._write
_op
_status
(nslcmop_id
, stage
)
4576 return error_detail_list
4579 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
4581 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
4582 The default-value is used. If it is between < > it look for a value at instantiation_params
4583 :param primitive_desc: portion of VNFD/NSD that describes primitive
4584 :param params: Params provided by user
4585 :param instantiation_params: Instantiation params provided by user
4586 :return: a dictionary with the calculated params
4588 calculated_params
= {}
4589 for parameter
in primitive_desc
.get("parameter", ()):
4590 param_name
= parameter
["name"]
4591 if param_name
in params
:
4592 calculated_params
[param_name
] = params
[param_name
]
4593 elif "default-value" in parameter
or "value" in parameter
:
4594 if "value" in parameter
:
4595 calculated_params
[param_name
] = parameter
["value"]
4597 calculated_params
[param_name
] = parameter
["default-value"]
4599 isinstance(calculated_params
[param_name
], str)
4600 and calculated_params
[param_name
].startswith("<")
4601 and calculated_params
[param_name
].endswith(">")
4603 if calculated_params
[param_name
][1:-1] in instantiation_params
:
4604 calculated_params
[param_name
] = instantiation_params
[
4605 calculated_params
[param_name
][1:-1]
4609 "Parameter {} needed to execute primitive {} not provided".format(
4610 calculated_params
[param_name
], primitive_desc
["name"]
4615 "Parameter {} needed to execute primitive {} not provided".format(
4616 param_name
, primitive_desc
["name"]
4620 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
4621 calculated_params
[param_name
] = yaml
.safe_dump(
4622 calculated_params
[param_name
], default_flow_style
=True, width
=256
4624 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[
4626 ].startswith("!!yaml "):
4627 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
4628 if parameter
.get("data-type") == "INTEGER":
4630 calculated_params
[param_name
] = int(calculated_params
[param_name
])
4631 except ValueError: # error converting string to int
4633 "Parameter {} of primitive {} must be integer".format(
4634 param_name
, primitive_desc
["name"]
4637 elif parameter
.get("data-type") == "BOOLEAN":
4638 calculated_params
[param_name
] = not (
4639 (str(calculated_params
[param_name
])).lower() == "false"
4642 # add always ns_config_info if primitive name is config
4643 if primitive_desc
["name"] == "config":
4644 if "ns_config_info" in instantiation_params
:
4645 calculated_params
["ns_config_info"] = instantiation_params
[
4648 return calculated_params
4650 def _look_for_deployed_vca(
4657 ee_descriptor_id
=None,
4659 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
4660 for vca
in deployed_vca
:
4663 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
4666 vdu_count_index
is not None
4667 and vdu_count_index
!= vca
["vdu_count_index"]
4670 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
4672 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
4676 # vca_deployed not found
4678 "charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
4679 " is not deployed".format(
4688 ee_id
= vca
.get("ee_id")
4690 "type", "lxc_proxy_charm"
4691 ) # default value for backward compatibility - proxy charm
4694 "charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
4695 "execution environment".format(
4696 member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
4699 return ee_id
, vca_type
4701 async def _ns_execute_primitive(
4707 retries_interval
=30,
4714 if primitive
== "config":
4715 primitive_params
= {"params": primitive_params
}
4717 vca_type
= vca_type
or "lxc_proxy_charm"
4721 output
= await asyncio
.wait_for(
4722 self
.vca_map
[vca_type
].exec_primitive(
4724 primitive_name
=primitive
,
4725 params_dict
=primitive_params
,
4726 progress_timeout
=self
.timeout_progress_primitive
,
4727 total_timeout
=self
.timeout_primitive
,
4732 timeout
=timeout
or self
.timeout_primitive
,
4736 except asyncio
.CancelledError
:
4738 except Exception as e
: # asyncio.TimeoutError
4739 if isinstance(e
, asyncio
.TimeoutError
):
4744 "Error executing action {} on {} -> {}".format(
4749 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
4751 return "FAILED", str(e
)
4753 return "COMPLETED", output
4755 except (LcmException
, asyncio
.CancelledError
):
4757 except Exception as e
:
4758 return "FAIL", "Error executing action {}: {}".format(primitive
, e
)
4760 async def vca_status_refresh(self
, nsr_id
, nslcmop_id
):
4762 Updating the vca_status with latest juju information in nsrs record
4763 :param: nsr_id: Id of the nsr
4764 :param: nslcmop_id: Id of the nslcmop
4768 self
.logger
.debug("Task ns={} action={} Enter".format(nsr_id
, nslcmop_id
))
4769 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4770 vca_id
= self
.get_vca_id({}, db_nsr
)
4771 if db_nsr
["_admin"]["deployed"]["K8s"]:
4772 for k8s_index
, k8s
in enumerate(db_nsr
["_admin"]["deployed"]["K8s"]):
4773 cluster_uuid
, kdu_instance
= k8s
["k8scluster-uuid"], k8s
["kdu-instance"]
4774 await self
._on
_update
_k
8s
_db
(
4775 cluster_uuid
, kdu_instance
, filter={"_id": nsr_id
}, vca_id
=vca_id
4778 for vca_index
, _
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
4779 table
, filter = "nsrs", {"_id": nsr_id
}
4780 path
= "_admin.deployed.VCA.{}.".format(vca_index
)
4781 await self
._on
_update
_n
2vc
_db
(table
, filter, path
, {})
4783 self
.logger
.debug("Task ns={} action={} Exit".format(nsr_id
, nslcmop_id
))
4784 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_vca_status_refresh")
4786 async def action(self
, nsr_id
, nslcmop_id
):
4787 # Try to lock HA task here
4788 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
4789 if not task_is_locked_by_me
:
4792 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
4793 self
.logger
.debug(logging_text
+ "Enter")
4794 # get all needed from database
4798 db_nslcmop_update
= {}
4799 nslcmop_operation_state
= None
4800 error_description_nslcmop
= None
4803 # wait for any previous tasks in process
4804 step
= "Waiting for previous operations to terminate"
4805 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
4807 self
._write
_ns
_status
(
4810 current_operation
="RUNNING ACTION",
4811 current_operation_id
=nslcmop_id
,
4814 step
= "Getting information from database"
4815 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
4816 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
4818 nsr_deployed
= db_nsr
["_admin"].get("deployed")
4819 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
4820 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
4821 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
4822 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
4823 primitive
= db_nslcmop
["operationParams"]["primitive"]
4824 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
4825 timeout_ns_action
= db_nslcmop
["operationParams"].get(
4826 "timeout_ns_action", self
.timeout_primitive
4830 step
= "Getting vnfr from database"
4831 db_vnfr
= self
.db
.get_one(
4832 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
4834 step
= "Getting vnfd from database"
4835 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
4837 # Sync filesystem before running a primitive
4838 self
.fs
.sync(db_vnfr
["vnfd-id"])
4840 step
= "Getting nsd from database"
4841 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
4843 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
4844 # for backward compatibility
4845 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
4846 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
4847 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
4848 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4850 # look for primitive
4851 config_primitive_desc
= descriptor_configuration
= None
4853 descriptor_configuration
= get_configuration(db_vnfd
, vdu_id
)
4855 descriptor_configuration
= get_configuration(db_vnfd
, kdu_name
)
4857 descriptor_configuration
= get_configuration(db_vnfd
, db_vnfd
["id"])
4859 descriptor_configuration
= db_nsd
.get("ns-configuration")
4861 if descriptor_configuration
and descriptor_configuration
.get(
4864 for config_primitive
in descriptor_configuration
["config-primitive"]:
4865 if config_primitive
["name"] == primitive
:
4866 config_primitive_desc
= config_primitive
4869 if not config_primitive_desc
:
4870 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
4872 "Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".format(
4876 primitive_name
= primitive
4877 ee_descriptor_id
= None
4879 primitive_name
= config_primitive_desc
.get(
4880 "execution-environment-primitive", primitive
4882 ee_descriptor_id
= config_primitive_desc
.get(
4883 "execution-environment-ref"
4889 (x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None
4891 desc_params
= parse_yaml_strings(vdur
.get("additionalParams"))
4894 (x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None
4896 desc_params
= parse_yaml_strings(kdur
.get("additionalParams"))
4898 desc_params
= parse_yaml_strings(
4899 db_vnfr
.get("additionalParamsForVnf")
4902 desc_params
= parse_yaml_strings(db_nsr
.get("additionalParamsForNs"))
4903 if kdu_name
and get_configuration(db_vnfd
, kdu_name
):
4904 kdu_configuration
= get_configuration(db_vnfd
, kdu_name
)
4906 for primitive
in kdu_configuration
.get("initial-config-primitive", []):
4907 actions
.add(primitive
["name"])
4908 for primitive
in kdu_configuration
.get("config-primitive", []):
4909 actions
.add(primitive
["name"])
4910 kdu_action
= True if primitive_name
in actions
else False
4912 # TODO check if ns is in a proper status
4914 primitive_name
in ("upgrade", "rollback", "status") or kdu_action
4916 # kdur and desc_params already set from before
4917 if primitive_params
:
4918 desc_params
.update(primitive_params
)
4919 # TODO Check if we will need something at vnf level
4920 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
4922 kdu_name
== kdu
["kdu-name"]
4923 and kdu
["member-vnf-index"] == vnf_index
4928 "KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
)
4931 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
4932 msg
= "unknown k8scluster-type '{}'".format(
4933 kdu
.get("k8scluster-type")
4935 raise LcmException(msg
)
4938 "collection": "nsrs",
4939 "filter": {"_id": nsr_id
},
4940 "path": "_admin.deployed.K8s.{}".format(index
),
4944 + "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
)
4946 step
= "Executing kdu {}".format(primitive_name
)
4947 if primitive_name
== "upgrade":
4948 if desc_params
.get("kdu_model"):
4949 kdu_model
= desc_params
.get("kdu_model")
4950 del desc_params
["kdu_model"]
4952 kdu_model
= kdu
.get("kdu-model")
4953 parts
= kdu_model
.split(sep
=":")
4955 kdu_model
= parts
[0]
4957 detailed_status
= await asyncio
.wait_for(
4958 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
4959 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4960 kdu_instance
=kdu
.get("kdu-instance"),
4962 kdu_model
=kdu_model
,
4965 timeout
=timeout_ns_action
,
4967 timeout
=timeout_ns_action
+ 10,
4970 logging_text
+ " Upgrade of kdu {} done".format(detailed_status
)
4972 elif primitive_name
== "rollback":
4973 detailed_status
= await asyncio
.wait_for(
4974 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
4975 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4976 kdu_instance
=kdu
.get("kdu-instance"),
4979 timeout
=timeout_ns_action
,
4981 elif primitive_name
== "status":
4982 detailed_status
= await asyncio
.wait_for(
4983 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
4984 cluster_uuid
=kdu
.get("k8scluster-uuid"),
4985 kdu_instance
=kdu
.get("kdu-instance"),
4988 timeout
=timeout_ns_action
,
4991 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(
4992 kdu
["kdu-name"], nsr_id
4994 params
= self
._map
_primitive
_params
(
4995 config_primitive_desc
, primitive_params
, desc_params
4998 detailed_status
= await asyncio
.wait_for(
4999 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
5000 cluster_uuid
=kdu
.get("k8scluster-uuid"),
5001 kdu_instance
=kdu_instance
,
5002 primitive_name
=primitive_name
,
5005 timeout
=timeout_ns_action
,
5008 timeout
=timeout_ns_action
,
5012 nslcmop_operation_state
= "COMPLETED"
5014 detailed_status
= ""
5015 nslcmop_operation_state
= "FAILED"
5017 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5018 nsr_deployed
["VCA"],
5019 member_vnf_index
=vnf_index
,
5021 vdu_count_index
=vdu_count_index
,
5022 ee_descriptor_id
=ee_descriptor_id
,
5024 for vca_index
, vca_deployed
in enumerate(
5025 db_nsr
["_admin"]["deployed"]["VCA"]
5027 if vca_deployed
.get("member-vnf-index") == vnf_index
:
5029 "collection": "nsrs",
5030 "filter": {"_id": nsr_id
},
5031 "path": "_admin.deployed.VCA.{}.".format(vca_index
),
5035 nslcmop_operation_state
,
5037 ) = await self
._ns
_execute
_primitive
(
5039 primitive
=primitive_name
,
5040 primitive_params
=self
._map
_primitive
_params
(
5041 config_primitive_desc
, primitive_params
, desc_params
5043 timeout
=timeout_ns_action
,
5049 db_nslcmop_update
["detailed-status"] = detailed_status
5050 error_description_nslcmop
= (
5051 detailed_status
if nslcmop_operation_state
== "FAILED" else ""
5055 + " task Done with result {} {}".format(
5056 nslcmop_operation_state
, detailed_status
5059 return # database update is called inside finally
5061 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
5062 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
5064 except asyncio
.CancelledError
:
5066 logging_text
+ "Cancelled Exception while '{}'".format(step
)
5068 exc
= "Operation was cancelled"
5069 except asyncio
.TimeoutError
:
5070 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
5072 except Exception as e
:
5073 exc
= traceback
.format_exc()
5074 self
.logger
.critical(
5075 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
5084 ) = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
5085 nslcmop_operation_state
= "FAILED"
5087 self
._write
_ns
_status
(
5091 ], # TODO check if degraded. For the moment use previous status
5092 current_operation
="IDLE",
5093 current_operation_id
=None,
5094 # error_description=error_description_nsr,
5095 # error_detail=error_detail,
5096 other_update
=db_nsr_update
,
5099 self
._write
_op
_status
(
5102 error_message
=error_description_nslcmop
,
5103 operation_state
=nslcmop_operation_state
,
5104 other_update
=db_nslcmop_update
,
5107 if nslcmop_operation_state
:
5109 await self
.msg
.aiowrite(
5114 "nslcmop_id": nslcmop_id
,
5115 "operationState": nslcmop_operation_state
,
5119 except Exception as e
:
5121 logging_text
+ "kafka_write notification Exception {}".format(e
)
5123 self
.logger
.debug(logging_text
+ "Exit")
5124 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
5125 return nslcmop_operation_state
, detailed_status
5127 async def scale(self
, nsr_id
, nslcmop_id
):
5128 # Try to lock HA task here
5129 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA("ns", "nslcmops", nslcmop_id
)
5130 if not task_is_locked_by_me
:
5133 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
5134 stage
= ["", "", ""]
5135 tasks_dict_info
= {}
5136 # ^ stage, step, VIM progress
5137 self
.logger
.debug(logging_text
+ "Enter")
5138 # get all needed from database
5140 db_nslcmop_update
= {}
5143 # in case of error, indicates what part of scale was failed to put nsr at error status
5144 scale_process
= None
5145 old_operational_status
= ""
5146 old_config_status
= ""
5149 # wait for any previous tasks in process
5150 step
= "Waiting for previous operations to terminate"
5151 await self
.lcm_tasks
.waitfor_related_HA("ns", "nslcmops", nslcmop_id
)
5152 self
._write
_ns
_status
(
5155 current_operation
="SCALING",
5156 current_operation_id
=nslcmop_id
,
5159 step
= "Getting nslcmop from database"
5161 step
+ " after having waited for previous tasks to be completed"
5163 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
5165 step
= "Getting nsr from database"
5166 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
5167 old_operational_status
= db_nsr
["operational-status"]
5168 old_config_status
= db_nsr
["config-status"]
5170 step
= "Parsing scaling parameters"
5171 db_nsr_update
["operational-status"] = "scaling"
5172 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5173 nsr_deployed
= db_nsr
["_admin"].get("deployed")
5175 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"][
5177 ]["member-vnf-index"]
5178 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"][
5180 ]["scaling-group-descriptor"]
5181 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
5182 # for backward compatibility
5183 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
5184 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
5185 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
5186 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5188 step
= "Getting vnfr from database"
5189 db_vnfr
= self
.db
.get_one(
5190 "vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
}
5193 vca_id
= self
.get_vca_id(db_vnfr
, db_nsr
)
5195 step
= "Getting vnfd from database"
5196 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
5198 base_folder
= db_vnfd
["_admin"]["storage"]
5200 step
= "Getting scaling-group-descriptor"
5201 scaling_descriptor
= find_in_list(
5202 get_scaling_aspect(db_vnfd
),
5203 lambda scale_desc
: scale_desc
["name"] == scaling_group
,
5205 if not scaling_descriptor
:
5207 "input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
5208 "at vnfd:scaling-group-descriptor".format(scaling_group
)
5211 step
= "Sending scale order to VIM"
5212 # TODO check if ns is in a proper status
5214 if not db_nsr
["_admin"].get("scaling-group"):
5219 "_admin.scaling-group": [
5220 {"name": scaling_group
, "nb-scale-op": 0}
5224 admin_scale_index
= 0
5226 for admin_scale_index
, admin_scale_info
in enumerate(
5227 db_nsr
["_admin"]["scaling-group"]
5229 if admin_scale_info
["name"] == scaling_group
:
5230 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
5232 else: # not found, set index one plus last element and add new entry with the name
5233 admin_scale_index
+= 1
5235 "_admin.scaling-group.{}.name".format(admin_scale_index
)
5238 vca_scaling_info
= []
5239 scaling_info
= {"scaling_group_name": scaling_group
, "vdu": [], "kdu": []}
5240 if scaling_type
== "SCALE_OUT":
5241 if "aspect-delta-details" not in scaling_descriptor
:
5243 "Aspect delta details not fount in scaling descriptor {}".format(
5244 scaling_descriptor
["name"]
5247 # count if max-instance-count is reached
5248 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5250 scaling_info
["scaling_direction"] = "OUT"
5251 scaling_info
["vdu-create"] = {}
5252 scaling_info
["kdu-create"] = {}
5253 for delta
in deltas
:
5254 for vdu_delta
in delta
.get("vdu-delta", {}):
5255 vdud
= get_vdu(db_vnfd
, vdu_delta
["id"])
5256 # vdu_index also provides the number of instance of the targeted vdu
5257 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5258 cloud_init_text
= self
._get
_vdu
_cloud
_init
_content
(
5262 additional_params
= (
5263 self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"])
5266 cloud_init_list
= []
5268 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5269 max_instance_count
= 10
5270 if vdu_profile
and "max-number-of-instances" in vdu_profile
:
5271 max_instance_count
= vdu_profile
.get(
5272 "max-number-of-instances", 10
5275 default_instance_num
= get_number_of_instances(
5278 instances_number
= vdu_delta
.get("number-of-instances", 1)
5279 nb_scale_op
+= instances_number
5281 new_instance_count
= nb_scale_op
+ default_instance_num
5282 # Control if new count is over max and vdu count is less than max.
5283 # Then assign new instance count
5284 if new_instance_count
> max_instance_count
> vdu_count
:
5285 instances_number
= new_instance_count
- max_instance_count
5287 instances_number
= instances_number
5289 if new_instance_count
> max_instance_count
:
5291 "reached the limit of {} (max-instance-count) "
5292 "scaling-out operations for the "
5293 "scaling-group-descriptor '{}'".format(
5294 nb_scale_op
, scaling_group
5297 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5299 # TODO Information of its own ip is not available because db_vnfr is not updated.
5300 additional_params
["OSM"] = get_osm_params(
5301 db_vnfr
, vdu_delta
["id"], vdu_index
+ x
5303 cloud_init_list
.append(
5304 self
._parse
_cloud
_init
(
5311 vca_scaling_info
.append(
5313 "osm_vdu_id": vdu_delta
["id"],
5314 "member-vnf-index": vnf_index
,
5316 "vdu_index": vdu_index
+ x
,
5319 scaling_info
["vdu-create"][vdu_delta
["id"]] = instances_number
5320 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5321 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5322 kdu_name
= kdu_profile
["kdu-name"]
5323 resource_name
= kdu_profile
["resource-name"]
5325 # Might have different kdus in the same delta
5326 # Should have list for each kdu
5327 if not scaling_info
["kdu-create"].get(kdu_name
, None):
5328 scaling_info
["kdu-create"][kdu_name
] = []
5330 kdur
= get_kdur(db_vnfr
, kdu_name
)
5331 if kdur
.get("helm-chart"):
5332 k8s_cluster_type
= "helm-chart-v3"
5333 self
.logger
.debug("kdur: {}".format(kdur
))
5335 kdur
.get("helm-version")
5336 and kdur
.get("helm-version") == "v2"
5338 k8s_cluster_type
= "helm-chart"
5339 raise NotImplementedError
5340 elif kdur
.get("juju-bundle"):
5341 k8s_cluster_type
= "juju-bundle"
5344 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5345 "juju-bundle. Maybe an old NBI version is running".format(
5346 db_vnfr
["member-vnf-index-ref"], kdu_name
5350 max_instance_count
= 10
5351 if kdu_profile
and "max-number-of-instances" in kdu_profile
:
5352 max_instance_count
= kdu_profile
.get(
5353 "max-number-of-instances", 10
5356 nb_scale_op
+= kdu_delta
.get("number-of-instances", 1)
5357 deployed_kdu
, _
= get_deployed_kdu(
5358 nsr_deployed
, kdu_name
, vnf_index
5360 if deployed_kdu
is None:
5362 "KDU '{}' for vnf '{}' not deployed".format(
5366 kdu_instance
= deployed_kdu
.get("kdu-instance")
5367 instance_num
= await self
.k8scluster_map
[
5369 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5370 kdu_replica_count
= instance_num
+ kdu_delta
.get(
5371 "number-of-instances", 1
5374 # Control if new count is over max and instance_num is less than max.
5375 # Then assign max instance number to kdu replica count
5376 if kdu_replica_count
> max_instance_count
> instance_num
:
5377 kdu_replica_count
= max_instance_count
5378 if kdu_replica_count
> max_instance_count
:
5380 "reached the limit of {} (max-instance-count) "
5381 "scaling-out operations for the "
5382 "scaling-group-descriptor '{}'".format(
5383 instance_num
, scaling_group
5387 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5388 vca_scaling_info
.append(
5390 "osm_kdu_id": kdu_name
,
5391 "member-vnf-index": vnf_index
,
5393 "kdu_index": instance_num
+ x
- 1,
5396 scaling_info
["kdu-create"][kdu_name
].append(
5398 "member-vnf-index": vnf_index
,
5400 "k8s-cluster-type": k8s_cluster_type
,
5401 "resource-name": resource_name
,
5402 "scale": kdu_replica_count
,
5405 elif scaling_type
== "SCALE_IN":
5406 deltas
= scaling_descriptor
.get("aspect-delta-details")["deltas"]
5408 scaling_info
["scaling_direction"] = "IN"
5409 scaling_info
["vdu-delete"] = {}
5410 scaling_info
["kdu-delete"] = {}
5412 for delta
in deltas
:
5413 for vdu_delta
in delta
.get("vdu-delta", {}):
5414 vdu_count
= vdu_index
= get_vdur_index(db_vnfr
, vdu_delta
)
5415 min_instance_count
= 0
5416 vdu_profile
= get_vdu_profile(db_vnfd
, vdu_delta
["id"])
5417 if vdu_profile
and "min-number-of-instances" in vdu_profile
:
5418 min_instance_count
= vdu_profile
["min-number-of-instances"]
5420 default_instance_num
= get_number_of_instances(
5421 db_vnfd
, vdu_delta
["id"]
5423 instance_num
= vdu_delta
.get("number-of-instances", 1)
5424 nb_scale_op
-= instance_num
5426 new_instance_count
= nb_scale_op
+ default_instance_num
5428 if new_instance_count
< min_instance_count
< vdu_count
:
5429 instances_number
= min_instance_count
- new_instance_count
5431 instances_number
= instance_num
5433 if new_instance_count
< min_instance_count
:
5435 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5436 "scaling-group-descriptor '{}'".format(
5437 nb_scale_op
, scaling_group
5440 for x
in range(vdu_delta
.get("number-of-instances", 1)):
5441 vca_scaling_info
.append(
5443 "osm_vdu_id": vdu_delta
["id"],
5444 "member-vnf-index": vnf_index
,
5446 "vdu_index": vdu_index
- 1 - x
,
5449 scaling_info
["vdu-delete"][vdu_delta
["id"]] = instances_number
5450 for kdu_delta
in delta
.get("kdu-resource-delta", {}):
5451 kdu_profile
= get_kdu_resource_profile(db_vnfd
, kdu_delta
["id"])
5452 kdu_name
= kdu_profile
["kdu-name"]
5453 resource_name
= kdu_profile
["resource-name"]
5455 if not scaling_info
["kdu-delete"].get(kdu_name
, None):
5456 scaling_info
["kdu-delete"][kdu_name
] = []
5458 kdur
= get_kdur(db_vnfr
, kdu_name
)
5459 if kdur
.get("helm-chart"):
5460 k8s_cluster_type
= "helm-chart-v3"
5461 self
.logger
.debug("kdur: {}".format(kdur
))
5463 kdur
.get("helm-version")
5464 and kdur
.get("helm-version") == "v2"
5466 k8s_cluster_type
= "helm-chart"
5467 raise NotImplementedError
5468 elif kdur
.get("juju-bundle"):
5469 k8s_cluster_type
= "juju-bundle"
5472 "kdu type for kdu='{}.{}' is neither helm-chart nor "
5473 "juju-bundle. Maybe an old NBI version is running".format(
5474 db_vnfr
["member-vnf-index-ref"], kdur
["kdu-name"]
5478 min_instance_count
= 0
5479 if kdu_profile
and "min-number-of-instances" in kdu_profile
:
5480 min_instance_count
= kdu_profile
["min-number-of-instances"]
5482 nb_scale_op
-= kdu_delta
.get("number-of-instances", 1)
5483 deployed_kdu
, _
= get_deployed_kdu(
5484 nsr_deployed
, kdu_name
, vnf_index
5486 if deployed_kdu
is None:
5488 "KDU '{}' for vnf '{}' not deployed".format(
5492 kdu_instance
= deployed_kdu
.get("kdu-instance")
5493 instance_num
= await self
.k8scluster_map
[
5495 ].get_scale_count(resource_name
, kdu_instance
, vca_id
=vca_id
)
5496 kdu_replica_count
= instance_num
- kdu_delta
.get(
5497 "number-of-instances", 1
5500 if kdu_replica_count
< min_instance_count
< instance_num
:
5501 kdu_replica_count
= min_instance_count
5502 if kdu_replica_count
< min_instance_count
:
5504 "reached the limit of {} (min-instance-count) scaling-in operations for the "
5505 "scaling-group-descriptor '{}'".format(
5506 instance_num
, scaling_group
5510 for x
in range(kdu_delta
.get("number-of-instances", 1)):
5511 vca_scaling_info
.append(
5513 "osm_kdu_id": kdu_name
,
5514 "member-vnf-index": vnf_index
,
5516 "kdu_index": instance_num
- x
- 1,
5519 scaling_info
["kdu-delete"][kdu_name
].append(
5521 "member-vnf-index": vnf_index
,
5523 "k8s-cluster-type": k8s_cluster_type
,
5524 "resource-name": resource_name
,
5525 "scale": kdu_replica_count
,
5529 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
5530 vdu_delete
= copy(scaling_info
.get("vdu-delete"))
5531 if scaling_info
["scaling_direction"] == "IN":
5532 for vdur
in reversed(db_vnfr
["vdur"]):
5533 if vdu_delete
.get(vdur
["vdu-id-ref"]):
5534 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
5535 scaling_info
["vdu"].append(
5537 "name": vdur
.get("name") or vdur
.get("vdu-name"),
5538 "vdu_id": vdur
["vdu-id-ref"],
5542 for interface
in vdur
["interfaces"]:
5543 scaling_info
["vdu"][-1]["interface"].append(
5545 "name": interface
["name"],
5546 "ip_address": interface
["ip-address"],
5547 "mac_address": interface
.get("mac-address"),
5550 # vdu_delete = vdu_scaling_info.pop("vdu-delete")
5553 step
= "Executing pre-scale vnf-config-primitive"
5554 if scaling_descriptor
.get("scaling-config-action"):
5555 for scaling_config_action
in scaling_descriptor
[
5556 "scaling-config-action"
5559 scaling_config_action
.get("trigger") == "pre-scale-in"
5560 and scaling_type
== "SCALE_IN"
5562 scaling_config_action
.get("trigger") == "pre-scale-out"
5563 and scaling_type
== "SCALE_OUT"
5565 vnf_config_primitive
= scaling_config_action
[
5566 "vnf-config-primitive-name-ref"
5568 step
= db_nslcmop_update
[
5570 ] = "executing pre-scale scaling-config-action '{}'".format(
5571 vnf_config_primitive
5574 # look for primitive
5575 for config_primitive
in (
5576 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5577 ).get("config-primitive", ()):
5578 if config_primitive
["name"] == vnf_config_primitive
:
5582 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
5583 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
5584 "primitive".format(scaling_group
, vnf_config_primitive
)
5587 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5588 if db_vnfr
.get("additionalParamsForVnf"):
5589 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5591 scale_process
= "VCA"
5592 db_nsr_update
["config-status"] = "configuring pre-scaling"
5593 primitive_params
= self
._map
_primitive
_params
(
5594 config_primitive
, {}, vnfr_params
5597 # Pre-scale retry check: Check if this sub-operation has been executed before
5598 op_index
= self
._check
_or
_add
_scale
_suboperation
(
5601 vnf_config_primitive
,
5605 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
5606 # Skip sub-operation
5607 result
= "COMPLETED"
5608 result_detail
= "Done"
5611 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
5612 vnf_config_primitive
, result
, result_detail
5616 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
5617 # New sub-operation: Get index of this sub-operation
5619 len(db_nslcmop
.get("_admin", {}).get("operations"))
5624 + "vnf_config_primitive={} New sub-operation".format(
5625 vnf_config_primitive
5629 # retry: Get registered params for this existing sub-operation
5630 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
5633 vnf_index
= op
.get("member_vnf_index")
5634 vnf_config_primitive
= op
.get("primitive")
5635 primitive_params
= op
.get("primitive_params")
5638 + "vnf_config_primitive={} Sub-operation retry".format(
5639 vnf_config_primitive
5642 # Execute the primitive, either with new (first-time) or registered (reintent) args
5643 ee_descriptor_id
= config_primitive
.get(
5644 "execution-environment-ref"
5646 primitive_name
= config_primitive
.get(
5647 "execution-environment-primitive", vnf_config_primitive
5649 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
5650 nsr_deployed
["VCA"],
5651 member_vnf_index
=vnf_index
,
5653 vdu_count_index
=None,
5654 ee_descriptor_id
=ee_descriptor_id
,
5656 result
, result_detail
= await self
._ns
_execute
_primitive
(
5665 + "vnf_config_primitive={} Done with result {} {}".format(
5666 vnf_config_primitive
, result
, result_detail
5669 # Update operationState = COMPLETED | FAILED
5670 self
._update
_suboperation
_status
(
5671 db_nslcmop
, op_index
, result
, result_detail
5674 if result
== "FAILED":
5675 raise LcmException(result_detail
)
5676 db_nsr_update
["config-status"] = old_config_status
5677 scale_process
= None
5681 "_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)
5684 "_admin.scaling-group.{}.time".format(admin_scale_index
)
5687 # SCALE-IN VCA - BEGIN
5688 if vca_scaling_info
:
5689 step
= db_nslcmop_update
[
5691 ] = "Deleting the execution environments"
5692 scale_process
= "VCA"
5693 for vca_info
in vca_scaling_info
:
5694 if vca_info
["type"] == "delete":
5695 member_vnf_index
= str(vca_info
["member-vnf-index"])
5697 logging_text
+ "vdu info: {}".format(vca_info
)
5699 if vca_info
.get("osm_vdu_id"):
5700 vdu_id
= vca_info
["osm_vdu_id"]
5701 vdu_index
= int(vca_info
["vdu_index"])
5704 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5705 member_vnf_index
, vdu_id
, vdu_index
5709 kdu_id
= vca_info
["osm_kdu_id"]
5712 ] = "Scaling member_vnf_index={}, kdu_id={}, vdu_index={} ".format(
5713 member_vnf_index
, kdu_id
, vdu_index
5715 stage
[2] = step
= "Scaling in VCA"
5716 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5717 vca_update
= db_nsr
["_admin"]["deployed"]["VCA"]
5718 config_update
= db_nsr
["configurationStatus"]
5719 for vca_index
, vca
in enumerate(vca_update
):
5721 (vca
or vca
.get("ee_id"))
5722 and vca
["member-vnf-index"] == member_vnf_index
5723 and vca
["vdu_count_index"] == vdu_index
5725 if vca
.get("vdu_id"):
5726 config_descriptor
= get_configuration(
5727 db_vnfd
, vca
.get("vdu_id")
5729 elif vca
.get("kdu_name"):
5730 config_descriptor
= get_configuration(
5731 db_vnfd
, vca
.get("kdu_name")
5734 config_descriptor
= get_configuration(
5735 db_vnfd
, db_vnfd
["id"]
5737 operation_params
= (
5738 db_nslcmop
.get("operationParams") or {}
5740 exec_terminate_primitives
= not operation_params
.get(
5741 "skip_terminate_primitives"
5742 ) and vca
.get("needed_terminate")
5743 task
= asyncio
.ensure_future(
5752 exec_primitives
=exec_terminate_primitives
,
5756 timeout
=self
.timeout_charm_delete
,
5759 tasks_dict_info
[task
] = "Terminating VCA {}".format(
5762 del vca_update
[vca_index
]
5763 del config_update
[vca_index
]
5764 # wait for pending tasks of terminate primitives
5768 + "Waiting for tasks {}".format(
5769 list(tasks_dict_info
.keys())
5772 error_list
= await self
._wait
_for
_tasks
(
5776 self
.timeout_charm_delete
, self
.timeout_ns_terminate
5781 tasks_dict_info
.clear()
5783 raise LcmException("; ".join(error_list
))
5785 db_vca_and_config_update
= {
5786 "_admin.deployed.VCA": vca_update
,
5787 "configurationStatus": config_update
,
5790 "nsrs", db_nsr
["_id"], db_vca_and_config_update
5792 scale_process
= None
5793 # SCALE-IN VCA - END
5796 if scaling_info
.get("vdu-create") or scaling_info
.get("vdu-delete"):
5797 scale_process
= "RO"
5798 if self
.ro_config
.get("ng"):
5799 await self
._scale
_ng
_ro
(
5800 logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, scaling_info
, stage
5802 scaling_info
.pop("vdu-create", None)
5803 scaling_info
.pop("vdu-delete", None)
5805 scale_process
= None
5809 if scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete"):
5810 scale_process
= "KDU"
5811 await self
._scale
_kdu
(
5812 logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
5814 scaling_info
.pop("kdu-create", None)
5815 scaling_info
.pop("kdu-delete", None)
5817 scale_process
= None
5821 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
5823 # SCALE-UP VCA - BEGIN
5824 if vca_scaling_info
:
5825 step
= db_nslcmop_update
[
5827 ] = "Creating new execution environments"
5828 scale_process
= "VCA"
5829 for vca_info
in vca_scaling_info
:
5830 if vca_info
["type"] == "create":
5831 member_vnf_index
= str(vca_info
["member-vnf-index"])
5833 logging_text
+ "vdu info: {}".format(vca_info
)
5835 vnfd_id
= db_vnfr
["vnfd-ref"]
5836 if vca_info
.get("osm_vdu_id"):
5837 vdu_index
= int(vca_info
["vdu_index"])
5838 deploy_params
= {"OSM": get_osm_params(db_vnfr
)}
5839 if db_vnfr
.get("additionalParamsForVnf"):
5840 deploy_params
.update(
5842 db_vnfr
["additionalParamsForVnf"].copy()
5845 descriptor_config
= get_configuration(
5846 db_vnfd
, db_vnfd
["id"]
5848 if descriptor_config
:
5853 logging_text
=logging_text
5854 + "member_vnf_index={} ".format(member_vnf_index
),
5857 nslcmop_id
=nslcmop_id
,
5863 member_vnf_index
=member_vnf_index
,
5864 vdu_index
=vdu_index
,
5866 deploy_params
=deploy_params
,
5867 descriptor_config
=descriptor_config
,
5868 base_folder
=base_folder
,
5869 task_instantiation_info
=tasks_dict_info
,
5872 vdu_id
= vca_info
["osm_vdu_id"]
5873 vdur
= find_in_list(
5874 db_vnfr
["vdur"], lambda vdu
: vdu
["vdu-id-ref"] == vdu_id
5876 descriptor_config
= get_configuration(db_vnfd
, vdu_id
)
5877 if vdur
.get("additionalParams"):
5878 deploy_params_vdu
= parse_yaml_strings(
5879 vdur
["additionalParams"]
5882 deploy_params_vdu
= deploy_params
5883 deploy_params_vdu
["OSM"] = get_osm_params(
5884 db_vnfr
, vdu_id
, vdu_count_index
=vdu_index
5886 if descriptor_config
:
5891 ] = "Scaling member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5892 member_vnf_index
, vdu_id
, vdu_index
5894 stage
[2] = step
= "Scaling out VCA"
5895 self
._write
_op
_status
(op_id
=nslcmop_id
, stage
=stage
)
5897 logging_text
=logging_text
5898 + "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
5899 member_vnf_index
, vdu_id
, vdu_index
5903 nslcmop_id
=nslcmop_id
,
5909 member_vnf_index
=member_vnf_index
,
5910 vdu_index
=vdu_index
,
5912 deploy_params
=deploy_params_vdu
,
5913 descriptor_config
=descriptor_config
,
5914 base_folder
=base_folder
,
5915 task_instantiation_info
=tasks_dict_info
,
5919 kdu_name
= vca_info
["osm_kdu_id"]
5920 descriptor_config
= get_configuration(db_vnfd
, kdu_name
)
5921 if descriptor_config
:
5923 kdu_index
= int(vca_info
["kdu_index"])
5927 for x
in db_vnfr
["kdur"]
5928 if x
["kdu-name"] == kdu_name
5930 deploy_params_kdu
= {"OSM": get_osm_params(db_vnfr
)}
5931 if kdur
.get("additionalParams"):
5932 deploy_params_kdu
= parse_yaml_strings(
5933 kdur
["additionalParams"]
5937 logging_text
=logging_text
,
5940 nslcmop_id
=nslcmop_id
,
5946 member_vnf_index
=member_vnf_index
,
5947 vdu_index
=kdu_index
,
5949 deploy_params
=deploy_params_kdu
,
5950 descriptor_config
=descriptor_config
,
5951 base_folder
=base_folder
,
5952 task_instantiation_info
=tasks_dict_info
,
5955 # SCALE-UP VCA - END
5956 scale_process
= None
5959 # execute primitive service POST-SCALING
5960 step
= "Executing post-scale vnf-config-primitive"
5961 if scaling_descriptor
.get("scaling-config-action"):
5962 for scaling_config_action
in scaling_descriptor
[
5963 "scaling-config-action"
5966 scaling_config_action
.get("trigger") == "post-scale-in"
5967 and scaling_type
== "SCALE_IN"
5969 scaling_config_action
.get("trigger") == "post-scale-out"
5970 and scaling_type
== "SCALE_OUT"
5972 vnf_config_primitive
= scaling_config_action
[
5973 "vnf-config-primitive-name-ref"
5975 step
= db_nslcmop_update
[
5977 ] = "executing post-scale scaling-config-action '{}'".format(
5978 vnf_config_primitive
5981 vnfr_params
= {"VDU_SCALE_INFO": scaling_info
}
5982 if db_vnfr
.get("additionalParamsForVnf"):
5983 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
5985 # look for primitive
5986 for config_primitive
in (
5987 get_configuration(db_vnfd
, db_vnfd
["id"]) or {}
5988 ).get("config-primitive", ()):
5989 if config_primitive
["name"] == vnf_config_primitive
:
5993 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
5994 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
5995 "config-primitive".format(
5996 scaling_group
, vnf_config_primitive
5999 scale_process
= "VCA"
6000 db_nsr_update
["config-status"] = "configuring post-scaling"
6001 primitive_params
= self
._map
_primitive
_params
(
6002 config_primitive
, {}, vnfr_params
6005 # Post-scale retry check: Check if this sub-operation has been executed before
6006 op_index
= self
._check
_or
_add
_scale
_suboperation
(
6009 vnf_config_primitive
,
6013 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
6014 # Skip sub-operation
6015 result
= "COMPLETED"
6016 result_detail
= "Done"
6019 + "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
6020 vnf_config_primitive
, result
, result_detail
6024 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
6025 # New sub-operation: Get index of this sub-operation
6027 len(db_nslcmop
.get("_admin", {}).get("operations"))
6032 + "vnf_config_primitive={} New sub-operation".format(
6033 vnf_config_primitive
6037 # retry: Get registered params for this existing sub-operation
6038 op
= db_nslcmop
.get("_admin", {}).get("operations", [])[
6041 vnf_index
= op
.get("member_vnf_index")
6042 vnf_config_primitive
= op
.get("primitive")
6043 primitive_params
= op
.get("primitive_params")
6046 + "vnf_config_primitive={} Sub-operation retry".format(
6047 vnf_config_primitive
6050 # Execute the primitive, either with new (first-time) or registered (reintent) args
6051 ee_descriptor_id
= config_primitive
.get(
6052 "execution-environment-ref"
6054 primitive_name
= config_primitive
.get(
6055 "execution-environment-primitive", vnf_config_primitive
6057 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(
6058 nsr_deployed
["VCA"],
6059 member_vnf_index
=vnf_index
,
6061 vdu_count_index
=None,
6062 ee_descriptor_id
=ee_descriptor_id
,
6064 result
, result_detail
= await self
._ns
_execute
_primitive
(
6073 + "vnf_config_primitive={} Done with result {} {}".format(
6074 vnf_config_primitive
, result
, result_detail
6077 # Update operationState = COMPLETED | FAILED
6078 self
._update
_suboperation
_status
(
6079 db_nslcmop
, op_index
, result
, result_detail
6082 if result
== "FAILED":
6083 raise LcmException(result_detail
)
6084 db_nsr_update
["config-status"] = old_config_status
6085 scale_process
= None
6090 ] = "" # "scaled {} {}".format(scaling_group, scaling_type)
6091 db_nsr_update
["operational-status"] = (
6093 if old_operational_status
== "failed"
6094 else old_operational_status
6096 db_nsr_update
["config-status"] = old_config_status
6099 ROclient
.ROClientException
,
6104 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
6106 except asyncio
.CancelledError
:
6108 logging_text
+ "Cancelled Exception while '{}'".format(step
)
6110 exc
= "Operation was cancelled"
6111 except Exception as e
:
6112 exc
= traceback
.format_exc()
6113 self
.logger
.critical(
6114 logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
),
6118 self
._write
_ns
_status
(
6121 current_operation
="IDLE",
6122 current_operation_id
=None,
6125 stage
[1] = "Waiting for instantiate pending tasks."
6126 self
.logger
.debug(logging_text
+ stage
[1])
6127 exc
= await self
._wait
_for
_tasks
(
6130 self
.timeout_ns_deploy
,
6138 ] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
6139 nslcmop_operation_state
= "FAILED"
6141 db_nsr_update
["operational-status"] = old_operational_status
6142 db_nsr_update
["config-status"] = old_config_status
6143 db_nsr_update
["detailed-status"] = ""
6145 if "VCA" in scale_process
:
6146 db_nsr_update
["config-status"] = "failed"
6147 if "RO" in scale_process
:
6148 db_nsr_update
["operational-status"] = "failed"
6151 ] = "FAILED scaling nslcmop={} {}: {}".format(
6152 nslcmop_id
, step
, exc
6155 error_description_nslcmop
= None
6156 nslcmop_operation_state
= "COMPLETED"
6157 db_nslcmop_update
["detailed-status"] = "Done"
6159 self
._write
_op
_status
(
6162 error_message
=error_description_nslcmop
,
6163 operation_state
=nslcmop_operation_state
,
6164 other_update
=db_nslcmop_update
,
6167 self
._write
_ns
_status
(
6170 current_operation
="IDLE",
6171 current_operation_id
=None,
6172 other_update
=db_nsr_update
,
6175 if nslcmop_operation_state
:
6179 "nslcmop_id": nslcmop_id
,
6180 "operationState": nslcmop_operation_state
,
6182 await self
.msg
.aiowrite("ns", "scaled", msg
, loop
=self
.loop
)
6183 except Exception as e
:
6185 logging_text
+ "kafka_write notification Exception {}".format(e
)
6187 self
.logger
.debug(logging_text
+ "Exit")
6188 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
6190 async def _scale_kdu(
6191 self
, logging_text
, nsr_id
, nsr_deployed
, db_vnfd
, vca_id
, scaling_info
6193 _scaling_info
= scaling_info
.get("kdu-create") or scaling_info
.get("kdu-delete")
6194 for kdu_name
in _scaling_info
:
6195 for kdu_scaling_info
in _scaling_info
[kdu_name
]:
6196 deployed_kdu
, index
= get_deployed_kdu(
6197 nsr_deployed
, kdu_name
, kdu_scaling_info
["member-vnf-index"]
6199 cluster_uuid
= deployed_kdu
["k8scluster-uuid"]
6200 kdu_instance
= deployed_kdu
["kdu-instance"]
6201 scale
= int(kdu_scaling_info
["scale"])
6202 k8s_cluster_type
= kdu_scaling_info
["k8s-cluster-type"]
6205 "collection": "nsrs",
6206 "filter": {"_id": nsr_id
},
6207 "path": "_admin.deployed.K8s.{}".format(index
),
6210 step
= "scaling application {}".format(
6211 kdu_scaling_info
["resource-name"]
6213 self
.logger
.debug(logging_text
+ step
)
6215 if kdu_scaling_info
["type"] == "delete":
6216 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6219 and kdu_config
.get("terminate-config-primitive")
6220 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6222 terminate_config_primitive_list
= kdu_config
.get(
6223 "terminate-config-primitive"
6225 terminate_config_primitive_list
.sort(
6226 key
=lambda val
: int(val
["seq"])
6230 terminate_config_primitive
6231 ) in terminate_config_primitive_list
:
6232 primitive_params_
= self
._map
_primitive
_params
(
6233 terminate_config_primitive
, {}, {}
6235 step
= "execute terminate config primitive"
6236 self
.logger
.debug(logging_text
+ step
)
6237 await asyncio
.wait_for(
6238 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6239 cluster_uuid
=cluster_uuid
,
6240 kdu_instance
=kdu_instance
,
6241 primitive_name
=terminate_config_primitive
["name"],
6242 params
=primitive_params_
,
6249 await asyncio
.wait_for(
6250 self
.k8scluster_map
[k8s_cluster_type
].scale(
6253 kdu_scaling_info
["resource-name"],
6256 timeout
=self
.timeout_vca_on_error
,
6259 if kdu_scaling_info
["type"] == "create":
6260 kdu_config
= get_configuration(db_vnfd
, kdu_name
)
6263 and kdu_config
.get("initial-config-primitive")
6264 and get_juju_ee_ref(db_vnfd
, kdu_name
) is None
6266 initial_config_primitive_list
= kdu_config
.get(
6267 "initial-config-primitive"
6269 initial_config_primitive_list
.sort(
6270 key
=lambda val
: int(val
["seq"])
6273 for initial_config_primitive
in initial_config_primitive_list
:
6274 primitive_params_
= self
._map
_primitive
_params
(
6275 initial_config_primitive
, {}, {}
6277 step
= "execute initial config primitive"
6278 self
.logger
.debug(logging_text
+ step
)
6279 await asyncio
.wait_for(
6280 self
.k8scluster_map
[k8s_cluster_type
].exec_primitive(
6281 cluster_uuid
=cluster_uuid
,
6282 kdu_instance
=kdu_instance
,
6283 primitive_name
=initial_config_primitive
["name"],
6284 params
=primitive_params_
,
6291 async def _scale_ng_ro(
6292 self
, logging_text
, db_nsr
, db_nslcmop
, db_vnfr
, vdu_scaling_info
, stage
6294 nsr_id
= db_nslcmop
["nsInstanceId"]
6295 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
6298 # read from db: vnfd's for every vnf
6301 # for each vnf in ns, read vnfd
6302 for vnfr
in self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
}):
6303 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
6304 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
6305 # if we haven't this vnfd, read it from db
6306 if not find_in_list(db_vnfds
, lambda a_vnfd
: a_vnfd
["id"] == vnfd_id
):
6308 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
6309 db_vnfds
.append(vnfd
)
6310 n2vc_key
= self
.n2vc
.get_public_key()
6311 n2vc_key_list
= [n2vc_key
]
6314 vdu_scaling_info
.get("vdu-create"),
6315 vdu_scaling_info
.get("vdu-delete"),
6318 # db_vnfr has been updated, update db_vnfrs to use it
6319 db_vnfrs
[db_vnfr
["member-vnf-index-ref"]] = db_vnfr
6320 await self
._instantiate
_ng
_ro
(
6330 start_deploy
=time(),
6331 timeout_ns_deploy
=self
.timeout_ns_deploy
,
6333 if vdu_scaling_info
.get("vdu-delete"):
6335 db_vnfr
, None, vdu_scaling_info
["vdu-delete"], mark_delete
=False
6338 async def add_prometheus_metrics(
6339 self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
6341 if not self
.prometheus
:
6343 # look if exist a file called 'prometheus*.j2' and
6344 artifact_content
= self
.fs
.dir_ls(artifact_path
)
6348 for f
in artifact_content
6349 if f
.startswith("prometheus") and f
.endswith(".j2")
6355 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
6359 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
6360 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
6362 vnfr_id
= vnfr_id
.replace("-", "")
6364 "JOB_NAME": vnfr_id
,
6365 "TARGET_IP": target_ip
,
6366 "EXPORTER_POD_IP": host_name
,
6367 "EXPORTER_POD_PORT": host_port
,
6369 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
6370 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
6371 for job
in job_list
:
6373 not isinstance(job
.get("job_name"), str)
6374 or vnfr_id
not in job
["job_name"]
6376 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
6377 job
["nsr_id"] = nsr_id
6378 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
6379 if await self
.prometheus
.update(job_dict
):
6380 return list(job_dict
.keys())
6382 def get_vca_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6384 Get VCA Cloud and VCA Cloud Credentials for the VIM account
6386 :param: vim_account_id: VIM Account ID
6388 :return: (cloud_name, cloud_credential)
6390 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6391 return config
.get("vca_cloud"), config
.get("vca_cloud_credential")
6393 def get_vca_k8s_cloud_and_credentials(self
, vim_account_id
: str) -> (str, str):
6395 Get VCA K8s Cloud and VCA K8s Cloud Credentials for the VIM account
6397 :param: vim_account_id: VIM Account ID
6399 :return: (cloud_name, cloud_credential)
6401 config
= VimAccountDB
.get_vim_account_with_id(vim_account_id
).get("config", {})
6402 return config
.get("vca_k8s_cloud"), config
.get("vca_k8s_cloud_credential")