1 # -*- coding: utf-8 -*-
4 # Copyright 2018 Telefonica S.A.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
22 import logging
.handlers
25 from jinja2
import Environment
, TemplateError
, TemplateNotFound
, StrictUndefined
, UndefinedError
27 from osm_lcm
import ROclient
28 from osm_lcm
.ng_ro
import NgRoClient
, NgRoException
29 from osm_lcm
.lcm_utils
import LcmException
, LcmExceptionNoMgmtIP
, LcmBase
, deep_get
, get_iterable
, populate_dict
30 from n2vc
.k8s_helm_conn
import K8sHelmConnector
31 from n2vc
.k8s_juju_conn
import K8sJujuConnector
33 from osm_common
.dbbase
import DbException
34 from osm_common
.fsbase
import FsException
36 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
37 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
39 from osm_lcm
.lcm_helm_conn
import LCMHelmConn
41 from copy
import copy
, deepcopy
42 from http
import HTTPStatus
44 from uuid
import uuid4
46 from random
import randint
48 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
51 class N2VCJujuConnectorLCM(N2VCJujuConnector
):
53 async def create_execution_environment(self
, namespace
: str, db_dict
: dict, reuse_ee_id
: str = None,
54 progress_timeout
: float = None, total_timeout
: float = None,
55 config
: dict = None, artifact_path
: str = None,
56 vca_type
: str = None) -> (str, dict):
57 # admit two new parameters, artifact_path and vca_type
58 if vca_type
== "k8s_proxy_charm":
59 ee_id
= await self
.install_k8s_proxy_charm(
60 charm_name
=artifact_path
[artifact_path
.rfind("/") + 1:],
62 artifact_path
=artifact_path
,
66 return await super().create_execution_environment(
67 namespace
=namespace
, db_dict
=db_dict
, reuse_ee_id
=reuse_ee_id
,
68 progress_timeout
=progress_timeout
, total_timeout
=total_timeout
)
70 async def install_configuration_sw(self
, ee_id
: str, artifact_path
: str, db_dict
: dict,
71 progress_timeout
: float = None, total_timeout
: float = None,
72 config
: dict = None, num_units
: int = 1, vca_type
: str = "lxc_proxy_charm"):
73 if vca_type
== "k8s_proxy_charm":
75 return await super().install_configuration_sw(
76 ee_id
=ee_id
, artifact_path
=artifact_path
, db_dict
=db_dict
, progress_timeout
=progress_timeout
,
77 total_timeout
=total_timeout
, config
=config
, num_units
=num_units
)
81 timeout_vca_on_error
= 5 * 60 # Time for charm from first time at blocked,error status to mark as failed
82 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
83 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
84 timeout_charm_delete
= 10 * 60
85 timeout_primitive
= 30 * 60 # timeout for primitive execution
86 timeout_progress_primitive
= 10 * 60 # timeout for some progress in a primitive execution
88 SUBOPERATION_STATUS_NOT_FOUND
= -1
89 SUBOPERATION_STATUS_NEW
= -2
90 SUBOPERATION_STATUS_SKIP
= -3
91 task_name_deploy_vca
= "Deploying VCA"
93 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
, prometheus
=None):
95 Init, Connect to database, filesystem storage, and messaging
96 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
103 logger
=logging
.getLogger('lcm.ns')
107 self
.lcm_tasks
= lcm_tasks
108 self
.timeout
= config
["timeout"]
109 self
.ro_config
= config
["ro_config"]
110 self
.ng_ro
= config
["ro_config"].get("ng")
111 self
.vca_config
= config
["VCA"].copy()
113 # create N2VC connector
114 self
.n2vc
= N2VCJujuConnectorLCM(
119 url
='{}:{}'.format(self
.vca_config
['host'], self
.vca_config
['port']),
120 username
=self
.vca_config
.get('user', None),
121 vca_config
=self
.vca_config
,
122 on_update_db
=self
._on
_update
_n
2vc
_db
125 self
.conn_helm_ee
= LCMHelmConn(
132 vca_config
=self
.vca_config
,
133 on_update_db
=self
._on
_update
_n
2vc
_db
136 self
.k8sclusterhelm
= K8sHelmConnector(
137 kubectl_command
=self
.vca_config
.get("kubectlpath"),
138 helm_command
=self
.vca_config
.get("helmpath"),
145 self
.k8sclusterjuju
= K8sJujuConnector(
146 kubectl_command
=self
.vca_config
.get("kubectlpath"),
147 juju_command
=self
.vca_config
.get("jujupath"),
154 self
.k8scluster_map
= {
155 "helm-chart": self
.k8sclusterhelm
,
156 "chart": self
.k8sclusterhelm
,
157 "juju-bundle": self
.k8sclusterjuju
,
158 "juju": self
.k8sclusterjuju
,
162 "lxc_proxy_charm": self
.n2vc
,
163 "native_charm": self
.n2vc
,
164 "k8s_proxy_charm": self
.n2vc
,
165 "helm": self
.conn_helm_ee
168 self
.prometheus
= prometheus
172 self
.RO
= NgRoClient(self
.loop
, **self
.ro_config
)
174 self
.RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
176 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
178 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
181 # TODO filter RO descriptor fields...
185 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
186 db_dict
['deploymentStatus'] = ro_descriptor
187 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
189 except Exception as e
:
190 self
.logger
.warn('Cannot write database RO deployment for ns={} -> {}'.format(nsrs_id
, e
))
192 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
):
194 # remove last dot from path (if exists)
195 if path
.endswith('.'):
198 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
199 # .format(table, filter, path, updated_data))
203 nsr_id
= filter.get('_id')
205 # read ns record from database
206 nsr
= self
.db
.get_one(table
='nsrs', q_filter
=filter)
207 current_ns_status
= nsr
.get('nsState')
209 # get vca status for NS
210 status_dict
= await self
.n2vc
.get_status(namespace
='.' + nsr_id
, yaml_format
=False)
214 db_dict
['vcaStatus'] = status_dict
216 # update configurationStatus for this VCA
218 vca_index
= int(path
[path
.rfind(".")+1:])
220 vca_list
= deep_get(target_dict
=nsr
, key_list
=('_admin', 'deployed', 'VCA'))
221 vca_status
= vca_list
[vca_index
].get('status')
223 configuration_status_list
= nsr
.get('configurationStatus')
224 config_status
= configuration_status_list
[vca_index
].get('status')
226 if config_status
== 'BROKEN' and vca_status
!= 'failed':
227 db_dict
['configurationStatus'][vca_index
] = 'READY'
228 elif config_status
!= 'BROKEN' and vca_status
== 'failed':
229 db_dict
['configurationStatus'][vca_index
] = 'BROKEN'
230 except Exception as e
:
231 # not update configurationStatus
232 self
.logger
.debug('Error updating vca_index (ignore): {}'.format(e
))
234 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
235 # if nsState = 'DEGRADED' check if all is OK
237 if current_ns_status
in ('READY', 'DEGRADED'):
238 error_description
= ''
240 if status_dict
.get('machines'):
241 for machine_id
in status_dict
.get('machines'):
242 machine
= status_dict
.get('machines').get(machine_id
)
243 # check machine agent-status
244 if machine
.get('agent-status'):
245 s
= machine
.get('agent-status').get('status')
248 error_description
+= 'machine {} agent-status={} ; '.format(machine_id
, s
)
249 # check machine instance status
250 if machine
.get('instance-status'):
251 s
= machine
.get('instance-status').get('status')
254 error_description
+= 'machine {} instance-status={} ; '.format(machine_id
, s
)
256 if status_dict
.get('applications'):
257 for app_id
in status_dict
.get('applications'):
258 app
= status_dict
.get('applications').get(app_id
)
259 # check application status
260 if app
.get('status'):
261 s
= app
.get('status').get('status')
264 error_description
+= 'application {} status={} ; '.format(app_id
, s
)
266 if error_description
:
267 db_dict
['errorDescription'] = error_description
268 if current_ns_status
== 'READY' and is_degraded
:
269 db_dict
['nsState'] = 'DEGRADED'
270 if current_ns_status
== 'DEGRADED' and not is_degraded
:
271 db_dict
['nsState'] = 'READY'
274 self
.update_db_2("nsrs", nsr_id
, db_dict
)
276 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
278 except Exception as e
:
279 self
.logger
.warn('Error updating NS state for ns={}: {}'.format(nsr_id
, e
))
282 def _parse_cloud_init(cloud_init_text
, additional_params
, vnfd_id
, vdu_id
):
284 env
= Environment(undefined
=StrictUndefined
)
285 template
= env
.from_string(cloud_init_text
)
286 return template
.render(additional_params
or {})
287 except UndefinedError
as e
:
288 raise LcmException("Variable {} at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
289 "file, must be provided in the instantiation parameters inside the "
290 "'additionalParamsForVnf/Vdu' block".format(e
, vnfd_id
, vdu_id
))
291 except (TemplateError
, TemplateNotFound
) as e
:
292 raise LcmException("Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".
293 format(vnfd_id
, vdu_id
, e
))
295 def _get_cloud_init(self
, vdu
, vnfd
):
297 cloud_init_content
= cloud_init_file
= None
298 if vdu
.get("cloud-init-file"):
299 base_folder
= vnfd
["_admin"]["storage"]
300 cloud_init_file
= "{}/{}/cloud_init/{}".format(base_folder
["folder"], base_folder
["pkg-dir"],
301 vdu
["cloud-init-file"])
302 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
303 cloud_init_content
= ci_file
.read()
304 elif vdu
.get("cloud-init"):
305 cloud_init_content
= vdu
["cloud-init"]
307 return cloud_init_content
308 except FsException
as e
:
309 raise LcmException("Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".
310 format(vnfd
["id"], vdu
["id"], cloud_init_file
, e
))
312 def _get_osm_params(self
, db_vnfr
, vdu_id
=None, vdu_count_index
=0):
313 osm_params
= {x
.replace("-", "_"): db_vnfr
[x
] for x
in ("ip-address", "vim-account-id", "vnfd-id", "vnfd-ref")
314 if db_vnfr
.get(x
) is not None}
315 osm_params
["ns_id"] = db_vnfr
["nsr-id-ref"]
316 osm_params
["vnf_id"] = db_vnfr
["_id"]
317 osm_params
["member_vnf_index"] = db_vnfr
["member-vnf-index-ref"]
318 if db_vnfr
.get("vdur"):
319 osm_params
["vdu"] = {}
320 for vdur
in db_vnfr
["vdur"]:
322 "count_index": vdur
["count-index"],
323 "vdu_id": vdur
["vdu-id-ref"],
326 if vdur
.get("ip-address"):
327 vdu
["ip_address"] = vdur
["ip-address"]
328 for iface
in vdur
["interfaces"]:
329 vdu
["interfaces"][iface
["name"]] = \
330 {x
.replace("-", "_"): iface
[x
] for x
in ("mac-address", "ip-address", "vnf-vld-id", "name")
331 if iface
.get(x
) is not None}
332 vdu_id_index
= "{}-{}".format(vdur
["vdu-id-ref"], vdur
["count-index"])
333 osm_params
["vdu"][vdu_id_index
] = vdu
335 osm_params
["vdu_id"] = vdu_id
336 osm_params
["count_index"] = vdu_count_index
339 def _get_vdu_additional_params(self
, db_vnfr
, vdu_id
):
340 vdur
= next(vdur
for vdur
in db_vnfr
.get("vdur") if vdu_id
== vdur
["vdu-id-ref"])
341 additional_params
= vdur
.get("additionalParams")
342 return self
._format
_additional
_params
(additional_params
)
344 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
346 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
347 :param vnfd: input vnfd
348 :param new_id: overrides vnf id if provided
349 :param additionalParams: Instantiation params for VNFs provided
350 :param nsrId: Id of the NSR
351 :return: copy of vnfd
353 vnfd_RO
= deepcopy(vnfd
)
354 # remove unused by RO configuration, monitoring, scaling and internal keys
355 vnfd_RO
.pop("_id", None)
356 vnfd_RO
.pop("_admin", None)
357 vnfd_RO
.pop("vnf-configuration", None)
358 vnfd_RO
.pop("monitoring-param", None)
359 vnfd_RO
.pop("scaling-group-descriptor", None)
360 vnfd_RO
.pop("kdu", None)
361 vnfd_RO
.pop("k8s-cluster", None)
363 vnfd_RO
["id"] = new_id
365 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
366 for vdu
in get_iterable(vnfd_RO
, "vdu"):
367 vdu
.pop("cloud-init-file", None)
368 vdu
.pop("cloud-init", None)
371 def _ns_params_2_RO(self
, ns_params
, nsd
, vnfd_dict
, db_vnfrs
, n2vc_key_list
):
373 Creates a RO ns descriptor from OSM ns_instantiate params
374 :param ns_params: OSM instantiate params
375 :param vnfd_dict: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
376 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index. {member-vnf-index: {vnfr_object}, ...}
377 :return: The RO ns descriptor
381 # TODO feature 1417: Check that no instantiation is set over PDU
382 # check if PDU forces a concrete vim-network-id and add it
383 # check if PDU contains a SDN-assist info (dpid, switch, port) and pass it to RO
385 def vim_account_2_RO(vim_account
):
386 if vim_account
in vim_2_RO
:
387 return vim_2_RO
[vim_account
]
389 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
390 if db_vim
["_admin"]["operationalState"] != "ENABLED":
391 raise LcmException("VIM={} is not available. operationalState={}".format(
392 vim_account
, db_vim
["_admin"]["operationalState"]))
393 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
394 vim_2_RO
[vim_account
] = RO_vim_id
397 def wim_account_2_RO(wim_account
):
398 if isinstance(wim_account
, str):
399 if wim_account
in wim_2_RO
:
400 return wim_2_RO
[wim_account
]
402 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
403 if db_wim
["_admin"]["operationalState"] != "ENABLED":
404 raise LcmException("WIM={} is not available. operationalState={}".format(
405 wim_account
, db_wim
["_admin"]["operationalState"]))
406 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
407 wim_2_RO
[wim_account
] = RO_wim_id
412 def ip_profile_2_RO(ip_profile
):
413 RO_ip_profile
= deepcopy((ip_profile
))
414 if "dns-server" in RO_ip_profile
:
415 if isinstance(RO_ip_profile
["dns-server"], list):
416 RO_ip_profile
["dns-address"] = []
417 for ds
in RO_ip_profile
.pop("dns-server"):
418 RO_ip_profile
["dns-address"].append(ds
['address'])
420 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
421 if RO_ip_profile
.get("ip-version") == "ipv4":
422 RO_ip_profile
["ip-version"] = "IPv4"
423 if RO_ip_profile
.get("ip-version") == "ipv6":
424 RO_ip_profile
["ip-version"] = "IPv6"
425 if "dhcp-params" in RO_ip_profile
:
426 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
432 # "name": ns_params["nsName"],
433 # "description": ns_params.get("nsDescription"),
434 "datacenter": vim_account_2_RO(ns_params
["vimAccountId"]),
435 "wim_account": wim_account_2_RO(ns_params
.get("wimAccountId")),
436 # "scenario": ns_params["nsdId"],
438 # set vim_account of each vnf if different from general vim_account.
439 # Get this information from <vnfr> database content, key vim-account-id
440 # Vim account can be set by placement_engine and it may be different from
441 # the instantiate parameters (vnfs.member-vnf-index.datacenter).
442 for vnf_index
, vnfr
in db_vnfrs
.items():
443 if vnfr
.get("vim-account-id") and vnfr
["vim-account-id"] != ns_params
["vimAccountId"]:
444 populate_dict(RO_ns_params
, ("vnfs", vnf_index
, "datacenter"), vim_account_2_RO(vnfr
["vim-account-id"]))
446 n2vc_key_list
= n2vc_key_list
or []
447 for vnfd_ref
, vnfd
in vnfd_dict
.items():
448 vdu_needed_access
= []
450 if vnfd
.get("vnf-configuration"):
451 ssh_required
= deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required"))
452 if ssh_required
and vnfd
.get("mgmt-interface"):
453 if vnfd
["mgmt-interface"].get("vdu-id"):
454 vdu_needed_access
.append(vnfd
["mgmt-interface"]["vdu-id"])
455 elif vnfd
["mgmt-interface"].get("cp"):
456 mgmt_cp
= vnfd
["mgmt-interface"]["cp"]
458 for vdu
in vnfd
.get("vdu", ()):
459 if vdu
.get("vdu-configuration"):
460 ssh_required
= deep_get(vdu
, ("vdu-configuration", "config-access", "ssh-access", "required"))
462 vdu_needed_access
.append(vdu
["id"])
464 for vdu_interface
in vdu
.get("interface"):
465 if vdu_interface
.get("external-connection-point-ref") and \
466 vdu_interface
["external-connection-point-ref"] == mgmt_cp
:
467 vdu_needed_access
.append(vdu
["id"])
471 if vdu_needed_access
:
472 for vnf_member
in nsd
.get("constituent-vnfd"):
473 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
475 for vdu
in vdu_needed_access
:
476 populate_dict(RO_ns_params
,
477 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
, "mgmt_keys"),
480 for vdu
in get_iterable(vnfd
, "vdu"):
481 cloud_init_text
= self
._get
_cloud
_init
(vdu
, vnfd
)
482 if not cloud_init_text
:
484 for vnf_member
in nsd
.get("constituent-vnfd"):
485 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
487 db_vnfr
= db_vnfrs
[vnf_member
["member-vnf-index"]]
488 additional_params
= self
._get
_vdu
_additional
_params
(db_vnfr
, vdu
["id"]) or {}
491 for vdu_index
in range(0, int(vdu
.get("count", 1))):
492 additional_params
["OSM"] = self
._get
_osm
_params
(db_vnfr
, vdu
["id"], vdu_index
)
493 cloud_init_list
.append(self
._parse
_cloud
_init
(cloud_init_text
, additional_params
, vnfd
["id"],
495 populate_dict(RO_ns_params
,
496 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
["id"], "cloud_init"),
499 if ns_params
.get("vduImage"):
500 RO_ns_params
["vduImage"] = ns_params
["vduImage"]
502 if ns_params
.get("ssh_keys"):
503 RO_ns_params
["cloud-config"] = {"key-pairs": ns_params
["ssh_keys"]}
504 for vnf_params
in get_iterable(ns_params
, "vnf"):
505 for constituent_vnfd
in nsd
["constituent-vnfd"]:
506 if constituent_vnfd
["member-vnf-index"] == vnf_params
["member-vnf-index"]:
507 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
510 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index={} is not present at nsd:"
511 "constituent-vnfd".format(vnf_params
["member-vnf-index"]))
513 for vdu_params
in get_iterable(vnf_params
, "vdu"):
514 # TODO feature 1417: check that this VDU exist and it is not a PDU
515 if vdu_params
.get("volume"):
516 for volume_params
in vdu_params
["volume"]:
517 if volume_params
.get("vim-volume-id"):
518 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
519 vdu_params
["id"], "devices", volume_params
["name"], "vim_id"),
520 volume_params
["vim-volume-id"])
521 if vdu_params
.get("interface"):
522 for interface_params
in vdu_params
["interface"]:
523 if interface_params
.get("ip-address"):
524 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
525 vdu_params
["id"], "interfaces", interface_params
["name"],
527 interface_params
["ip-address"])
528 if interface_params
.get("mac-address"):
529 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
530 vdu_params
["id"], "interfaces", interface_params
["name"],
532 interface_params
["mac-address"])
533 if interface_params
.get("floating-ip-required"):
534 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
535 vdu_params
["id"], "interfaces", interface_params
["name"],
537 interface_params
["floating-ip-required"])
539 for internal_vld_params
in get_iterable(vnf_params
, "internal-vld"):
540 if internal_vld_params
.get("vim-network-name"):
541 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
542 internal_vld_params
["name"], "vim-network-name"),
543 internal_vld_params
["vim-network-name"])
544 if internal_vld_params
.get("vim-network-id"):
545 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
546 internal_vld_params
["name"], "vim-network-id"),
547 internal_vld_params
["vim-network-id"])
548 if internal_vld_params
.get("ip-profile"):
549 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
550 internal_vld_params
["name"], "ip-profile"),
551 ip_profile_2_RO(internal_vld_params
["ip-profile"]))
552 if internal_vld_params
.get("provider-network"):
554 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
555 internal_vld_params
["name"], "provider-network"),
556 internal_vld_params
["provider-network"].copy())
558 for icp_params
in get_iterable(internal_vld_params
, "internal-connection-point"):
561 for vdu_descriptor
in vnf_descriptor
["vdu"]:
562 for vdu_interface
in vdu_descriptor
["interface"]:
563 if vdu_interface
.get("internal-connection-point-ref") == icp_params
["id-ref"]:
564 if icp_params
.get("ip-address"):
565 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
566 vdu_descriptor
["id"], "interfaces",
567 vdu_interface
["name"], "ip_address"),
568 icp_params
["ip-address"])
570 if icp_params
.get("mac-address"):
571 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
572 vdu_descriptor
["id"], "interfaces",
573 vdu_interface
["name"], "mac_address"),
574 icp_params
["mac-address"])
580 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index[{}]:"
581 "internal-vld:id-ref={} is not present at vnfd:internal-"
582 "connection-point".format(vnf_params
["member-vnf-index"],
583 icp_params
["id-ref"]))
585 for vld_params
in get_iterable(ns_params
, "vld"):
586 if "ip-profile" in vld_params
:
587 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "ip-profile"),
588 ip_profile_2_RO(vld_params
["ip-profile"]))
590 if vld_params
.get("provider-network"):
592 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "provider-network"),
593 vld_params
["provider-network"].copy())
595 if "wimAccountId" in vld_params
and vld_params
["wimAccountId"] is not None:
596 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "wim_account"),
597 wim_account_2_RO(vld_params
["wimAccountId"])),
598 if vld_params
.get("vim-network-name"):
600 if isinstance(vld_params
["vim-network-name"], dict):
601 for vim_account
, vim_net
in vld_params
["vim-network-name"].items():
602 RO_vld_sites
.append({
603 "netmap-use": vim_net
,
604 "datacenter": vim_account_2_RO(vim_account
)
606 else: # isinstance str
607 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-name"]})
609 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
611 if vld_params
.get("vim-network-id"):
613 if isinstance(vld_params
["vim-network-id"], dict):
614 for vim_account
, vim_net
in vld_params
["vim-network-id"].items():
615 RO_vld_sites
.append({
616 "netmap-use": vim_net
,
617 "datacenter": vim_account_2_RO(vim_account
)
619 else: # isinstance str
620 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-id"]})
622 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
623 if vld_params
.get("ns-net"):
624 if isinstance(vld_params
["ns-net"], dict):
625 for vld_id
, instance_scenario_id
in vld_params
["ns-net"].items():
626 RO_vld_ns_net
= {"instance_scenario_id": instance_scenario_id
, "osm_id": vld_id
}
627 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "use-network"), RO_vld_ns_net
)
628 if "vnfd-connection-point-ref" in vld_params
:
629 for cp_params
in vld_params
["vnfd-connection-point-ref"]:
631 for constituent_vnfd
in nsd
["constituent-vnfd"]:
632 if constituent_vnfd
["member-vnf-index"] == cp_params
["member-vnf-index-ref"]:
633 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
637 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={} "
638 "is not present at nsd:constituent-vnfd".format(cp_params
["member-vnf-index-ref"]))
640 for vdu_descriptor
in vnf_descriptor
["vdu"]:
641 for interface_descriptor
in vdu_descriptor
["interface"]:
642 if interface_descriptor
.get("external-connection-point-ref") == \
643 cp_params
["vnfd-connection-point-ref"]:
650 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={}:"
651 "vnfd-connection-point-ref={} is not present at vnfd={}".format(
652 cp_params
["member-vnf-index-ref"],
653 cp_params
["vnfd-connection-point-ref"],
654 vnf_descriptor
["id"]))
655 if cp_params
.get("ip-address"):
656 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
657 vdu_descriptor
["id"], "interfaces",
658 interface_descriptor
["name"], "ip_address"),
659 cp_params
["ip-address"])
660 if cp_params
.get("mac-address"):
661 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
662 vdu_descriptor
["id"], "interfaces",
663 interface_descriptor
["name"], "mac_address"),
664 cp_params
["mac-address"])
667 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None):
668 # make a copy to do not change
669 vdu_create
= copy(vdu_create
)
670 vdu_delete
= copy(vdu_delete
)
672 vdurs
= db_vnfr
.get("vdur")
675 vdu_index
= len(vdurs
)
678 vdur
= vdurs
[vdu_index
]
679 if vdur
.get("pdu-type"):
681 vdu_id_ref
= vdur
["vdu-id-ref"]
682 if vdu_create
and vdu_create
.get(vdu_id_ref
):
683 vdur_copy
= deepcopy(vdur
)
684 vdur_copy
["status"] = "BUILD"
685 vdur_copy
["status-detailed"] = None
686 vdur_copy
["ip_address"]: None
687 for iface
in vdur_copy
["interfaces"]:
688 iface
["ip-address"] = None
689 iface
["mac-address"] = None
690 iface
.pop("mgmt_vnf", None) # only first vdu can be managment of vnf # TODO ALF
691 for index
in range(0, vdu_create
[vdu_id_ref
]):
692 vdur_copy
["_id"] = str(uuid4())
693 vdur_copy
["count-index"] += 1
694 vdurs
.insert(vdu_index
+1+index
, vdur_copy
)
695 self
.logger
.debug("scale out, adding vdu={}".format(vdur_copy
))
696 vdur_copy
= deepcopy(vdur_copy
)
698 del vdu_create
[vdu_id_ref
]
699 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
701 vdu_delete
[vdu_id_ref
] -= 1
702 if not vdu_delete
[vdu_id_ref
]:
703 del vdu_delete
[vdu_id_ref
]
704 # check all operations are done
705 if vdu_create
or vdu_delete
:
706 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
709 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
712 vnfr_update
= {"vdur": vdurs
}
713 db_vnfr
["vdur"] = vdurs
714 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
716 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
718 Updates database nsr with the RO info for the created vld
719 :param ns_update_nsr: dictionary to be filled with the updated info
720 :param db_nsr: content of db_nsr. This is also modified
721 :param nsr_desc_RO: nsr descriptor from RO
722 :return: Nothing, LcmException is raised on errors
725 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
726 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
727 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
729 vld
["vim-id"] = net_RO
.get("vim_net_id")
730 vld
["name"] = net_RO
.get("vim_name")
731 vld
["status"] = net_RO
.get("status")
732 vld
["status-detailed"] = net_RO
.get("error_msg")
733 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
736 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
738 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
740 for db_vnfr
in db_vnfrs
.values():
741 vnfr_update
= {"status": "ERROR"}
742 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
743 if "status" not in vdur
:
744 vdur
["status"] = "ERROR"
745 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
747 vdur
["status-detailed"] = str(error_text
)
748 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
749 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
750 except DbException
as e
:
751 self
.logger
.error("Cannot update vnf. {}".format(e
))
753 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
755 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
756 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
757 :param nsr_desc_RO: nsr descriptor from RO
758 :return: Nothing, LcmException is raised on errors
760 for vnf_index
, db_vnfr
in db_vnfrs
.items():
761 for vnf_RO
in nsr_desc_RO
["vnfs"]:
762 if vnf_RO
["member_vnf_index"] != vnf_index
:
765 if vnf_RO
.get("ip_address"):
766 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
767 elif not db_vnfr
.get("ip-address"):
768 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
769 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
771 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
772 vdur_RO_count_index
= 0
773 if vdur
.get("pdu-type"):
775 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
776 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
778 if vdur
["count-index"] != vdur_RO_count_index
:
779 vdur_RO_count_index
+= 1
781 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
782 if vdur_RO
.get("ip_address"):
783 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
785 vdur
["ip-address"] = None
786 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
787 vdur
["name"] = vdur_RO
.get("vim_name")
788 vdur
["status"] = vdur_RO
.get("status")
789 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
790 for ifacer
in get_iterable(vdur
, "interfaces"):
791 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
792 if ifacer
["name"] == interface_RO
.get("internal_name"):
793 ifacer
["ip-address"] = interface_RO
.get("ip_address")
794 ifacer
["mac-address"] = interface_RO
.get("mac_address")
797 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
799 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
800 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
803 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
804 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
806 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
807 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
808 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
810 vld
["vim-id"] = net_RO
.get("vim_net_id")
811 vld
["name"] = net_RO
.get("vim_name")
812 vld
["status"] = net_RO
.get("status")
813 vld
["status-detailed"] = net_RO
.get("error_msg")
814 vnfr_update
["vld.{}".format(vld_index
)] = vld
817 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
818 vnf_index
, vld
["id"]))
820 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
824 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
826 def _get_ns_config_info(self
, nsr_id
):
828 Generates a mapping between vnf,vdu elements and the N2VC id
829 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
830 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
831 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
832 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
834 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
835 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
837 ns_config_info
= {"osm-config-mapping": mapping
}
838 for vca
in vca_deployed_list
:
839 if not vca
["member-vnf-index"]:
841 if not vca
["vdu_id"]:
842 mapping
[vca
["member-vnf-index"]] = vca
["application"]
844 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
846 return ns_config_info
849 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
, ee_descriptor_id
):
851 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
852 primitives as verify-ssh-credentials, or config when needed
853 :param desc_primitive_list: information of the descriptor
854 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
855 this element contains a ssh public key
856 :param ee_descriptor_id: execution environment descriptor id. It is the value of
857 XXX_configuration.execution-environment-list.INDEX.id; it can be None
858 :return: The modified list. Can ba an empty list, but always a list
861 primitive_list
= desc_primitive_list
or []
863 # filter primitives by ee_id
864 primitive_list
= [p
for p
in primitive_list
if p
.get("execution-environment-ref") == ee_descriptor_id
]
868 primitive_list
.sort(key
=lambda val
: int(val
['seq']))
870 # look for primitive config, and get the position. None if not present
871 config_position
= None
872 for index
, primitive
in enumerate(primitive_list
):
873 if primitive
["name"] == "config":
874 config_position
= index
877 # for NS, add always a config primitive if not present (bug 874)
878 if not vca_deployed
["member-vnf-index"] and config_position
is None:
879 primitive_list
.insert(0, {"name": "config", "parameter": []})
881 # TODO revise if needed: for VNF/VDU add verify-ssh-credentials after config
882 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
883 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
884 return primitive_list
886 async def _instantiate_ng_ro(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
887 n2vc_key_list
, stage
, start_deploy
, timeout_ns_deploy
):
888 nslcmop_id
= db_nslcmop
["_id"]
890 "name": db_nsr
["name"],
893 "image": deepcopy(db_nsr
["image"]),
894 "flavor": deepcopy(db_nsr
["flavor"]),
895 "action_id": nslcmop_id
,
897 for image
in target
["image"]:
898 image
["vim_info"] = []
899 for flavor
in target
["flavor"]:
900 flavor
["vim_info"] = []
902 ns_params
= db_nslcmop
.get("operationParams")
904 if ns_params
.get("ssh_keys"):
905 ssh_keys
+= ns_params
.get("ssh_keys")
907 ssh_keys
+= n2vc_key_list
910 for vld_index
, vld
in enumerate(nsd
.get("vld")):
911 target_vld
= {"id": vld
["id"],
913 "mgmt-network": vld
.get("mgmt-network", False),
914 "type": vld
.get("type"),
915 "vim_info": [{"vim-network-name": vld
.get("vim-network-name"),
916 "vim_account_id": ns_params
["vimAccountId"]}],
918 for cp
in vld
["vnfd-connection-point-ref"]:
919 cp2target
["member_vnf:{}.{}".format(cp
["member-vnf-index-ref"], cp
["vnfd-connection-point-ref"])] = \
920 "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
921 target
["ns"]["vld"].append(target_vld
)
922 for vnfr
in db_vnfrs
.values():
923 vnfd
= db_vnfds_ref
[vnfr
["vnfd-ref"]]
924 target_vnf
= deepcopy(vnfr
)
925 for vld
in target_vnf
.get("vld", ()):
926 # check if connected to a ns.vld
927 vnf_cp
= next((cp
for cp
in vnfd
.get("connection-point", ()) if
928 cp
.get("internal-vld-ref") == vld
["id"]), None)
930 ns_cp
= "member_vnf:{}.{}".format(vnfr
["member-vnf-index-ref"], vnf_cp
["id"])
931 if cp2target
.get(ns_cp
):
932 vld
["target"] = cp2target
[ns_cp
]
933 vld
["vim_info"] = [{"vim-network-name": vld
.get("vim-network-name"),
934 "vim_account_id": vnfr
["vim-account-id"]}]
936 for vdur
in target_vnf
.get("vdur", ()):
937 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
938 vdud_index
, vdud
= next(k
for k
in enumerate(vnfd
["vdu"]) if k
[1]["id"] == vdur
["vdu-id-ref"])
939 # vdur["additionalParams"] = vnfr.get("additionalParamsForVnf") # TODO additional params for VDU
942 if deep_get(vdud
, ("vdu-configuration", "config-access", "ssh-access", "required")):
943 vdur
["ssh-keys"] = ssh_keys
944 vdur
["ssh-access-required"] = True
945 elif deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required")) and \
946 any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"]):
947 vdur
["ssh-keys"] = ssh_keys
948 vdur
["ssh-access-required"] = True
951 if vdud
.get("cloud-init-file"):
952 vdur
["cloud-init"] = "{}:file:{}".format(vnfd
["_id"], vdud
.get("cloud-init-file"))
953 elif vdud
.get("cloud-init"):
954 vdur
["cloud-init"] = "{}:vdu:{}".format(vnfd
["_id"], vdud_index
)
957 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
958 if not next((vi
for vi
in ns_flavor
["vim_info"] if
959 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
960 ns_flavor
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
962 ns_image
= target
["image"][int(vdur
["ns-image-id"])]
963 if not next((vi
for vi
in ns_image
["vim_info"] if
964 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
965 ns_image
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
967 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
968 target
["vnf"].append(target_vnf
)
970 desc
= await self
.RO
.deploy(nsr_id
, target
)
971 action_id
= desc
["action_id"]
972 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
)
976 "_admin.deployed.RO.operational-status": "running",
977 "detailed-status": " ".join(stage
)
979 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
980 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
981 self
._write
_op
_status
(nslcmop_id
, stage
)
982 self
.logger
.debug(logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
))
985 async def _wait_ng_ro(self
, nsr_id
, action_id
, nslcmop_id
, start_time
, timeout
, stage
):
986 detailed_status_old
= None
988 while time() <= start_time
+ timeout
:
989 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
990 if desc_status
["status"] == "FAILED":
991 raise NgRoException(desc_status
["details"])
992 elif desc_status
["status"] == "BUILD":
993 stage
[2] = "VIM: ({})".format(desc_status
["details"])
994 elif desc_status
["status"] == "DONE":
995 stage
[2] = "Deployed at VIM"
998 assert False, "ROclient.check_ns_status returns unknown {}".format(desc_status
["status"])
999 if stage
[2] != detailed_status_old
:
1000 detailed_status_old
= stage
[2]
1001 db_nsr_update
["detailed-status"] = " ".join(stage
)
1002 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1003 self
._write
_op
_status
(nslcmop_id
, stage
)
1004 await asyncio
.sleep(5, loop
=self
.loop
)
1005 else: # timeout_ns_deploy
1006 raise NgRoException("Timeout waiting ns to deploy")
1008 async def _terminate_ng_ro(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
1012 start_deploy
= time()
1020 desc
= await self
.RO
.deploy(nsr_id
, target
)
1021 action_id
= desc
["action_id"]
1022 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
1023 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
1024 self
.logger
.debug(logging_text
+ "ns terminate action at RO. action_id={}".format(action_id
))
1027 delete_timeout
= 20 * 60 # 20 minutes
1028 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
)
1030 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1031 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1033 await self
.RO
.delete(nsr_id
)
1034 except Exception as e
:
1035 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
1036 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1037 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
1038 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
1039 self
.logger
.debug(logging_text
+ "RO_action_id={} already deleted".format(action_id
))
1040 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
1041 failed_detail
.append("delete conflict: {}".format(e
))
1042 self
.logger
.debug(logging_text
+ "RO_action_id={} delete conflict: {}".format(action_id
, e
))
1044 failed_detail
.append("delete error: {}".format(e
))
1045 self
.logger
.error(logging_text
+ "RO_action_id={} delete error: {}".format(action_id
, e
))
1048 stage
[2] = "Error deleting from VIM"
1050 stage
[2] = "Deleted from VIM"
1051 db_nsr_update
["detailed-status"] = " ".join(stage
)
1052 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1053 self
._write
_op
_status
(nslcmop_id
, stage
)
1056 raise LcmException("; ".join(failed_detail
))
1059 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
1060 n2vc_key_list
, stage
):
1063 :param logging_text: preffix text to use at logging
1064 :param nsr_id: nsr identity
1065 :param nsd: database content of ns descriptor
1066 :param db_nsr: database content of ns record
1067 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1069 :param db_vnfds_ref: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1070 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1071 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1072 :return: None or exception
1076 RO_descriptor_number
= 0 # number of descriptors created at RO
1077 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
1078 nslcmop_id
= db_nslcmop
["_id"]
1079 start_deploy
= time()
1080 ns_params
= db_nslcmop
.get("operationParams")
1081 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1082 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1084 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1086 # Check for and optionally request placement optimization. Database will be updated if placement activated
1087 stage
[2] = "Waiting for Placement."
1088 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1089 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1090 for vnfr
in db_vnfrs
.values():
1091 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1094 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1097 return await self
._instantiate
_ng
_ro
(logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
,
1098 db_vnfds_ref
, n2vc_key_list
, stage
, start_deploy
,
1101 # get vnfds, instantiate at RO
1102 for c_vnf
in nsd
.get("constituent-vnfd", ()):
1103 member_vnf_index
= c_vnf
["member-vnf-index"]
1104 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
1105 vnfd_ref
= vnfd
["id"]
1107 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
1108 db_nsr_update
["detailed-status"] = " ".join(stage
)
1109 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1110 self
._write
_op
_status
(nslcmop_id
, stage
)
1112 # self.logger.debug(logging_text + stage[2])
1113 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
1114 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
1115 RO_descriptor_number
+= 1
1117 # look position at deployed.RO.vnfd if not present it will be appended at the end
1118 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
1119 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
1122 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
1123 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
1126 RO_update
= {"member-vnf-index": member_vnf_index
}
1127 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
1129 RO_update
["id"] = vnfd_list
[0]["uuid"]
1130 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
1131 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
1133 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
1134 get("additionalParamsForVnf"), nsr_id
)
1135 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
1136 RO_update
["id"] = desc
["uuid"]
1137 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
1138 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
1139 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
1140 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
1145 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1146 db_nsr_update
["detailed-status"] = " ".join(stage
)
1147 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1148 self
._write
_op
_status
(nslcmop_id
, stage
)
1150 # self.logger.debug(logging_text + stage[2])
1151 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
1152 RO_descriptor_number
+= 1
1153 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
1155 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
1156 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
1157 nsd_ref
, RO_nsd_uuid
))
1159 nsd_RO
= deepcopy(nsd
)
1160 nsd_RO
["id"] = RO_osm_nsd_id
1161 nsd_RO
.pop("_id", None)
1162 nsd_RO
.pop("_admin", None)
1163 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
1164 member_vnf_index
= c_vnf
["member-vnf-index"]
1165 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1166 for c_vld
in nsd_RO
.get("vld", ()):
1167 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
1168 member_vnf_index
= cp
["member-vnf-index-ref"]
1169 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1171 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
1172 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1173 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
1174 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
1175 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1178 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1179 db_nsr_update
["detailed-status"] = " ".join(stage
)
1180 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1181 self
._write
_op
_status
(nslcmop_id
, stage
)
1183 # if present use it unless in error status
1184 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
1187 stage
[2] = "Looking for existing ns at RO"
1188 db_nsr_update
["detailed-status"] = " ".join(stage
)
1189 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1190 self
._write
_op
_status
(nslcmop_id
, stage
)
1191 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
1192 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1194 except ROclient
.ROClientException
as e
:
1195 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
1197 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1199 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1200 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1201 if ns_status
== "ERROR":
1202 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
1203 self
.logger
.debug(logging_text
+ stage
[2])
1204 await self
.RO
.delete("ns", RO_nsr_id
)
1205 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1207 stage
[2] = "Checking dependencies"
1208 db_nsr_update
["detailed-status"] = " ".join(stage
)
1209 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1210 self
._write
_op
_status
(nslcmop_id
, stage
)
1211 # self.logger.debug(logging_text + stage[2])
1213 # check if VIM is creating and wait look if previous tasks in process
1214 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
1216 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
1217 self
.logger
.debug(logging_text
+ stage
[2])
1218 await asyncio
.wait(task_dependency
, timeout
=3600)
1219 if ns_params
.get("vnf"):
1220 for vnf
in ns_params
["vnf"]:
1221 if "vimAccountId" in vnf
:
1222 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
1223 vnf
["vimAccountId"])
1225 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
1226 self
.logger
.debug(logging_text
+ stage
[2])
1227 await asyncio
.wait(task_dependency
, timeout
=3600)
1229 stage
[2] = "Checking instantiation parameters."
1230 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, db_vnfrs
, n2vc_key_list
)
1231 stage
[2] = "Deploying ns at VIM."
1232 db_nsr_update
["detailed-status"] = " ".join(stage
)
1233 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1234 self
._write
_op
_status
(nslcmop_id
, stage
)
1236 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
1237 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
1238 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1239 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
1240 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
1242 # wait until NS is ready
1243 stage
[2] = "Waiting VIM to deploy ns."
1244 db_nsr_update
["detailed-status"] = " ".join(stage
)
1245 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1246 self
._write
_op
_status
(nslcmop_id
, stage
)
1247 detailed_status_old
= None
1248 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
1251 while time() <= start_deploy
+ timeout_ns_deploy
:
1252 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1255 if desc
!= old_desc
:
1256 # desc has changed => update db
1257 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
1260 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1261 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1262 if ns_status
== "ERROR":
1263 raise ROclient
.ROClientException(ns_status_info
)
1264 elif ns_status
== "BUILD":
1265 stage
[2] = "VIM: ({})".format(ns_status_info
)
1266 elif ns_status
== "ACTIVE":
1267 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
1269 self
.ns_update_vnfr(db_vnfrs
, desc
)
1271 except LcmExceptionNoMgmtIP
:
1274 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
1275 if stage
[2] != detailed_status_old
:
1276 detailed_status_old
= stage
[2]
1277 db_nsr_update
["detailed-status"] = " ".join(stage
)
1278 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1279 self
._write
_op
_status
(nslcmop_id
, stage
)
1280 await asyncio
.sleep(5, loop
=self
.loop
)
1281 else: # timeout_ns_deploy
1282 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
1285 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
1287 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
1288 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1289 stage
[2] = "Deployed at VIM"
1290 db_nsr_update
["detailed-status"] = " ".join(stage
)
1291 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1292 self
._write
_op
_status
(nslcmop_id
, stage
)
1293 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
1294 # self.logger.debug(logging_text + "Deployed at VIM")
1295 except (ROclient
.ROClientException
, LcmException
, DbException
, NgRoException
) as e
:
1296 stage
[2] = "ERROR deploying at VIM"
1297 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1300 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1302 Wait for kdu to be up, get ip address
1303 :param logging_text: prefix use for logging
1310 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1313 while nb_tries
< 360:
1314 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1315 kdur
= next((x
for x
in get_iterable(db_vnfr
, "kdur") if x
.get("kdu-name") == kdu_name
), None)
1317 raise LcmException("Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
))
1318 if kdur
.get("status"):
1319 if kdur
["status"] in ("READY", "ENABLED"):
1320 return kdur
.get("ip-address")
1322 raise LcmException("target KDU={} is in error state".format(kdu_name
))
1324 await asyncio
.sleep(10, loop
=self
.loop
)
1326 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1328 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
1330 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1331 :param logging_text: prefix use for logging
1336 :param pub_key: public ssh key to inject, None to skip
1337 :param user: user to apply the public ssh key
1341 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
1345 target_vdu_id
= None
1351 if ro_retries
>= 360: # 1 hour
1352 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
1354 await asyncio
.sleep(10, loop
=self
.loop
)
1357 if not target_vdu_id
:
1358 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1360 if not vdu_id
: # for the VNF case
1361 if db_vnfr
.get("status") == "ERROR":
1362 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1363 ip_address
= db_vnfr
.get("ip-address")
1366 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1368 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1369 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1371 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1372 vdur
= db_vnfr
["vdur"][0]
1374 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1377 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1378 ip_address
= vdur
.get("ip-address")
1381 target_vdu_id
= vdur
["vdu-id-ref"]
1382 elif vdur
.get("status") == "ERROR":
1383 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1385 if not target_vdu_id
:
1388 # inject public key into machine
1389 if pub_key
and user
:
1390 # wait until NS is deployed at RO
1392 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1393 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1397 # self.logger.debug(logging_text + "Inserting RO key")
1398 if vdur
.get("pdu-type"):
1399 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1402 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1404 target
= {"action": "inject_ssh_key", "key": pub_key
, "user": user
,
1405 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdu_id
}]}],
1407 await self
.RO
.deploy(nsr_id
, target
)
1409 result_dict
= await self
.RO
.create_action(
1411 item_id_name
=ro_nsr_id
,
1412 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1414 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1415 if not result_dict
or not isinstance(result_dict
, dict):
1416 raise LcmException("Unknown response from RO when injecting key")
1417 for result
in result_dict
.values():
1418 if result
.get("vim_result") == 200:
1421 raise ROclient
.ROClientException("error injecting key: {}".format(
1422 result
.get("description")))
1424 except NgRoException
as e
:
1425 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1426 except ROclient
.ROClientException
as e
:
1428 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1432 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1438 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1440 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1442 my_vca
= vca_deployed_list
[vca_index
]
1443 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1444 # vdu or kdu: no dependencies
1448 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1449 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1450 configuration_status_list
= db_nsr
["configurationStatus"]
1451 for index
, vca_deployed
in enumerate(configuration_status_list
):
1452 if index
== vca_index
:
1455 if not my_vca
.get("member-vnf-index") or \
1456 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1457 internal_status
= configuration_status_list
[index
].get("status")
1458 if internal_status
== 'READY':
1460 elif internal_status
== 'BROKEN':
1461 raise LcmException("Configuration aborted because dependent charm/s has failed")
1465 # no dependencies, return
1467 await asyncio
.sleep(10)
1470 raise LcmException("Configuration aborted because dependent charm/s timeout")
1472 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1473 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
, vca_type
, vca_name
,
1474 ee_config_descriptor
):
1475 nsr_id
= db_nsr
["_id"]
1476 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1477 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1478 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1479 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1481 'collection': 'nsrs',
1482 'filter': {'_id': nsr_id
},
1483 'path': db_update_entry
1489 element_under_configuration
= nsr_id
1493 vnfr_id
= db_vnfr
["_id"]
1494 osm_config
["osm"]["vnf_id"] = vnfr_id
1496 namespace
= "{nsi}.{ns}".format(
1497 nsi
=nsi_id
if nsi_id
else "",
1501 element_type
= 'VNF'
1502 element_under_configuration
= vnfr_id
1503 namespace
+= ".{}".format(vnfr_id
)
1505 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1506 element_type
= 'VDU'
1507 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1508 osm_config
["osm"]["vdu_id"] = vdu_id
1510 namespace
+= ".{}".format(kdu_name
)
1511 element_type
= 'KDU'
1512 element_under_configuration
= kdu_name
1513 osm_config
["osm"]["kdu_name"] = kdu_name
1516 artifact_path
= "{}/{}/{}/{}".format(
1517 base_folder
["folder"],
1518 base_folder
["pkg-dir"],
1519 "charms" if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") else "helm-charts",
1522 # get initial_config_primitive_list that applies to this element
1523 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1525 # add config if not present for NS charm
1526 ee_descriptor_id
= ee_config_descriptor
.get("id")
1527 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1528 vca_deployed
, ee_descriptor_id
)
1530 # n2vc_redesign STEP 3.1
1531 # find old ee_id if exists
1532 ee_id
= vca_deployed
.get("ee_id")
1534 # create or register execution environment in VCA
1535 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm"):
1537 self
._write
_configuration
_status
(
1539 vca_index
=vca_index
,
1541 element_under_configuration
=element_under_configuration
,
1542 element_type
=element_type
1545 step
= "create execution environment"
1546 self
.logger
.debug(logging_text
+ step
)
1547 ee_id
, credentials
= await self
.vca_map
[vca_type
].create_execution_environment(
1548 namespace
=namespace
,
1552 artifact_path
=artifact_path
,
1555 elif vca_type
== "native_charm":
1556 step
= "Waiting to VM being up and getting IP address"
1557 self
.logger
.debug(logging_text
+ step
)
1558 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1559 user
=None, pub_key
=None)
1560 credentials
= {"hostname": rw_mgmt_ip
}
1562 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1563 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1564 # merged. Meanwhile let's get username from initial-config-primitive
1565 if not username
and initial_config_primitive_list
:
1566 for config_primitive
in initial_config_primitive_list
:
1567 for param
in config_primitive
.get("parameter", ()):
1568 if param
["name"] == "ssh-username":
1569 username
= param
["value"]
1572 raise LcmException("Cannot determine the username neither with 'initial-config-primitive' nor with "
1573 "'config-access.ssh-access.default-user'")
1574 credentials
["username"] = username
1575 # n2vc_redesign STEP 3.2
1577 self
._write
_configuration
_status
(
1579 vca_index
=vca_index
,
1580 status
='REGISTERING',
1581 element_under_configuration
=element_under_configuration
,
1582 element_type
=element_type
1585 step
= "register execution environment {}".format(credentials
)
1586 self
.logger
.debug(logging_text
+ step
)
1587 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1588 credentials
=credentials
, namespace
=namespace
, db_dict
=db_dict
)
1590 # for compatibility with MON/POL modules, the need model and application name at database
1591 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1592 ee_id_parts
= ee_id
.split('.')
1593 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1594 if len(ee_id_parts
) >= 2:
1595 model_name
= ee_id_parts
[0]
1596 application_name
= ee_id_parts
[1]
1597 db_nsr_update
[db_update_entry
+ "model"] = model_name
1598 db_nsr_update
[db_update_entry
+ "application"] = application_name
1600 # n2vc_redesign STEP 3.3
1601 step
= "Install configuration Software"
1603 self
._write
_configuration
_status
(
1605 vca_index
=vca_index
,
1606 status
='INSTALLING SW',
1607 element_under_configuration
=element_under_configuration
,
1608 element_type
=element_type
,
1609 other_update
=db_nsr_update
1612 # TODO check if already done
1613 self
.logger
.debug(logging_text
+ step
)
1615 if vca_type
== "native_charm":
1616 config_primitive
= next((p
for p
in initial_config_primitive_list
if p
["name"] == "config"), None)
1617 if config_primitive
:
1618 config
= self
._map
_primitive
_params
(
1624 if vca_type
== "lxc_proxy_charm":
1625 if element_type
== "NS":
1626 num_units
= db_nsr
.get("config-units") or 1
1627 elif element_type
== "VNF":
1628 num_units
= db_vnfr
.get("config-units") or 1
1629 elif element_type
== "VDU":
1630 for v
in db_vnfr
["vdur"]:
1631 if vdu_id
== v
["vdu-id-ref"]:
1632 num_units
= v
.get("config-units") or 1
1635 await self
.vca_map
[vca_type
].install_configuration_sw(
1637 artifact_path
=artifact_path
,
1640 num_units
=num_units
,
1644 # write in db flag of configuration_sw already installed
1645 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1647 # add relations for this VCA (wait for other peers related with this VCA)
1648 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
,
1649 vca_index
=vca_index
, vca_type
=vca_type
)
1651 # if SSH access is required, then get execution environment SSH public
1652 # if native charm we have waited already to VM be UP
1653 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm"):
1656 # self.logger.debug("get ssh key block")
1657 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1658 # self.logger.debug("ssh key needed")
1659 # Needed to inject a ssh key
1660 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1661 step
= "Install configuration Software, getting public ssh key"
1662 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1664 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1666 # self.logger.debug("no need to get ssh key")
1667 step
= "Waiting to VM being up and getting IP address"
1668 self
.logger
.debug(logging_text
+ step
)
1670 # n2vc_redesign STEP 5.1
1671 # wait for RO (ip-address) Insert pub_key into VM
1674 rw_mgmt_ip
= await self
.wait_kdu_up(logging_text
, nsr_id
, vnfr_id
, kdu_name
)
1676 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
,
1677 vdu_index
, user
=user
, pub_key
=pub_key
)
1679 rw_mgmt_ip
= None # This is for a NS configuration
1681 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1683 # store rw_mgmt_ip in deploy params for later replacement
1684 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1686 # n2vc_redesign STEP 6 Execute initial config primitive
1687 step
= 'execute initial config primitive'
1689 # wait for dependent primitives execution (NS -> VNF -> VDU)
1690 if initial_config_primitive_list
:
1691 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1693 # stage, in function of element type: vdu, kdu, vnf or ns
1694 my_vca
= vca_deployed_list
[vca_index
]
1695 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1697 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1698 elif my_vca
.get("member-vnf-index"):
1700 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1703 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1705 self
._write
_configuration
_status
(
1707 vca_index
=vca_index
,
1708 status
='EXECUTING PRIMITIVE'
1711 self
._write
_op
_status
(
1716 check_if_terminated_needed
= True
1717 for initial_config_primitive
in initial_config_primitive_list
:
1718 # adding information on the vca_deployed if it is a NS execution environment
1719 if not vca_deployed
["member-vnf-index"]:
1720 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1721 # TODO check if already done
1722 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1724 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1725 self
.logger
.debug(logging_text
+ step
)
1726 await self
.vca_map
[vca_type
].exec_primitive(
1728 primitive_name
=initial_config_primitive
["name"],
1729 params_dict
=primitive_params_
,
1732 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1733 if check_if_terminated_needed
:
1734 if config_descriptor
.get('terminate-config-primitive'):
1735 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1736 check_if_terminated_needed
= False
1738 # TODO register in database that primitive is done
1740 # STEP 7 Configure metrics
1741 if vca_type
== "helm":
1742 prometheus_jobs
= await self
.add_prometheus_metrics(
1744 artifact_path
=artifact_path
,
1745 ee_config_descriptor
=ee_config_descriptor
,
1748 target_ip
=rw_mgmt_ip
,
1751 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "prometheus_jobs": prometheus_jobs
})
1753 step
= "instantiated at VCA"
1754 self
.logger
.debug(logging_text
+ step
)
1756 self
._write
_configuration
_status
(
1758 vca_index
=vca_index
,
1762 except Exception as e
: # TODO not use Exception but N2VC exception
1763 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1764 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1765 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1766 self
._write
_configuration
_status
(
1768 vca_index
=vca_index
,
1771 raise LcmException("{} {}".format(step
, e
)) from e
1773 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1774 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1776 Update db_nsr fields.
1779 :param current_operation:
1780 :param current_operation_id:
1781 :param error_description:
1782 :param error_detail:
1783 :param other_update: Other required changes at database if provided, will be cleared
1787 db_dict
= other_update
or {}
1788 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1789 db_dict
["_admin.current-operation"] = current_operation_id
1790 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1791 db_dict
["currentOperation"] = current_operation
1792 db_dict
["currentOperationID"] = current_operation_id
1793 db_dict
["errorDescription"] = error_description
1794 db_dict
["errorDetail"] = error_detail
1797 db_dict
["nsState"] = ns_state
1798 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1799 except DbException
as e
:
1800 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1802 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1803 operation_state
: str = None, other_update
: dict = None):
1805 db_dict
= other_update
or {}
1806 db_dict
['queuePosition'] = queuePosition
1807 if isinstance(stage
, list):
1808 db_dict
['stage'] = stage
[0]
1809 db_dict
['detailed-status'] = " ".join(stage
)
1810 elif stage
is not None:
1811 db_dict
['stage'] = str(stage
)
1813 if error_message
is not None:
1814 db_dict
['errorMessage'] = error_message
1815 if operation_state
is not None:
1816 db_dict
['operationState'] = operation_state
1817 db_dict
["statusEnteredTime"] = time()
1818 self
.update_db_2("nslcmops", op_id
, db_dict
)
1819 except DbException
as e
:
1820 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1822 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1824 nsr_id
= db_nsr
["_id"]
1825 # configurationStatus
1826 config_status
= db_nsr
.get('configurationStatus')
1828 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1829 enumerate(config_status
) if v
}
1831 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1833 except DbException
as e
:
1834 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1836 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1837 element_under_configuration
: str = None, element_type
: str = None,
1838 other_update
: dict = None):
1840 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1841 # .format(vca_index, status))
1844 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1845 db_dict
= other_update
or {}
1847 db_dict
[db_path
+ 'status'] = status
1848 if element_under_configuration
:
1849 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1851 db_dict
[db_path
+ 'elementType'] = element_type
1852 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1853 except DbException
as e
:
1854 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1855 .format(status
, nsr_id
, vca_index
, e
))
1857 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1859 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1860 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1861 Database is used because the result can be obtained from a different LCM worker in case of HA.
1862 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1863 :param db_nslcmop: database content of nslcmop
1864 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1865 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
1866 computed 'vim-account-id'
1869 nslcmop_id
= db_nslcmop
['_id']
1870 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1871 if placement_engine
== "PLA":
1872 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1873 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1874 db_poll_interval
= 5
1875 wait
= db_poll_interval
* 10
1877 while not pla_result
and wait
>= 0:
1878 await asyncio
.sleep(db_poll_interval
)
1879 wait
-= db_poll_interval
1880 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1881 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1884 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1886 for pla_vnf
in pla_result
['vnf']:
1887 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1888 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1891 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1893 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1896 def update_nsrs_with_pla_result(self
, params
):
1898 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1899 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1900 except Exception as e
:
1901 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1903 async def instantiate(self
, nsr_id
, nslcmop_id
):
1906 :param nsr_id: ns instance to deploy
1907 :param nslcmop_id: operation to run
1911 # Try to lock HA task here
1912 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1913 if not task_is_locked_by_me
:
1914 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1917 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1918 self
.logger
.debug(logging_text
+ "Enter")
1920 # get all needed from database
1922 # database nsrs record
1925 # database nslcmops record
1928 # update operation on nsrs
1930 # update operation on nslcmops
1931 db_nslcmop_update
= {}
1933 nslcmop_operation_state
= None
1934 db_vnfrs
= {} # vnf's info indexed by member-index
1936 tasks_dict_info
= {} # from task to info text
1939 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1940 # ^ stage, step, VIM progress
1942 # wait for any previous tasks in process
1943 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1945 stage
[1] = "Sync filesystem from database."
1946 self
.fs
.sync() # TODO, make use of partial sync, only for the needed packages
1948 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1949 stage
[1] = "Reading from database."
1950 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1951 db_nsr_update
["detailed-status"] = "creating"
1952 db_nsr_update
["operational-status"] = "init"
1953 self
._write
_ns
_status
(
1955 ns_state
="BUILDING",
1956 current_operation
="INSTANTIATING",
1957 current_operation_id
=nslcmop_id
,
1958 other_update
=db_nsr_update
1960 self
._write
_op
_status
(
1966 # read from db: operation
1967 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
1968 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1969 ns_params
= db_nslcmop
.get("operationParams")
1970 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1971 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1973 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1976 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
1977 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1978 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
1979 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
1981 # nsr_name = db_nsr["name"] # TODO short-name??
1983 # read from db: vnf's of this ns
1984 stage
[1] = "Getting vnfrs from db."
1985 self
.logger
.debug(logging_text
+ stage
[1])
1986 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1988 # read from db: vnfd's for every vnf
1989 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1990 db_vnfds
= {} # every vnfd data indexed by vnf id
1991 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1993 # for each vnf in ns, read vnfd
1994 for vnfr
in db_vnfrs_list
:
1995 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1996 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1997 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1999 # if we haven't this vnfd, read it from db
2000 if vnfd_id
not in db_vnfds
:
2002 stage
[1] = "Getting vnfd={} id='{}' from db.".format(vnfd_id
, vnfd_ref
)
2003 self
.logger
.debug(logging_text
+ stage
[1])
2004 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2007 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
2008 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
2009 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
2011 # Get or generates the _admin.deployed.VCA list
2012 vca_deployed_list
= None
2013 if db_nsr
["_admin"].get("deployed"):
2014 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
2015 if vca_deployed_list
is None:
2016 vca_deployed_list
= []
2017 configuration_status_list
= []
2018 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2019 db_nsr_update
["configurationStatus"] = configuration_status_list
2020 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
2021 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2022 elif isinstance(vca_deployed_list
, dict):
2023 # maintain backward compatibility. Change a dict to list at database
2024 vca_deployed_list
= list(vca_deployed_list
.values())
2025 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
2026 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
2028 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
2029 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
2030 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
2032 # set state to INSTANTIATED. When instantiated NBI will not delete directly
2033 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
2034 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2035 self
.db
.set_list("vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"})
2037 # n2vc_redesign STEP 2 Deploy Network Scenario
2038 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
2039 self
._write
_op
_status
(
2044 stage
[1] = "Deploying KDUs."
2045 # self.logger.debug(logging_text + "Before deploy_kdus")
2046 # Call to deploy_kdus in case exists the "vdu:kdu" param
2047 await self
.deploy_kdus(
2048 logging_text
=logging_text
,
2050 nslcmop_id
=nslcmop_id
,
2053 task_instantiation_info
=tasks_dict_info
,
2056 stage
[1] = "Getting VCA public key."
2057 # n2vc_redesign STEP 1 Get VCA public ssh-key
2058 # feature 1429. Add n2vc public key to needed VMs
2059 n2vc_key
= self
.n2vc
.get_public_key()
2060 n2vc_key_list
= [n2vc_key
]
2061 if self
.vca_config
.get("public_key"):
2062 n2vc_key_list
.append(self
.vca_config
["public_key"])
2064 stage
[1] = "Deploying NS at VIM."
2065 task_ro
= asyncio
.ensure_future(
2066 self
.instantiate_RO(
2067 logging_text
=logging_text
,
2071 db_nslcmop
=db_nslcmop
,
2073 db_vnfds_ref
=db_vnfds_ref
,
2074 n2vc_key_list
=n2vc_key_list
,
2078 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2079 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2081 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2082 stage
[1] = "Deploying Execution Environments."
2083 self
.logger
.debug(logging_text
+ stage
[1])
2085 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2086 # get_iterable() returns a value from a dict or empty tuple if key does not exist
2087 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
2088 vnfd_id
= c_vnf
["vnfd-id-ref"]
2089 vnfd
= db_vnfds_ref
[vnfd_id
]
2090 member_vnf_index
= str(c_vnf
["member-vnf-index"])
2091 db_vnfr
= db_vnfrs
[member_vnf_index
]
2092 base_folder
= vnfd
["_admin"]["storage"]
2098 # Get additional parameters
2099 deploy_params
= {"OSM": self
._get
_osm
_params
(db_vnfr
)}
2100 if db_vnfr
.get("additionalParamsForVnf"):
2101 deploy_params
.update(self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy()))
2103 descriptor_config
= vnfd
.get("vnf-configuration")
2104 if descriptor_config
:
2106 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
2109 nslcmop_id
=nslcmop_id
,
2115 member_vnf_index
=member_vnf_index
,
2116 vdu_index
=vdu_index
,
2118 deploy_params
=deploy_params
,
2119 descriptor_config
=descriptor_config
,
2120 base_folder
=base_folder
,
2121 task_instantiation_info
=tasks_dict_info
,
2125 # Deploy charms for each VDU that supports one.
2126 for vdud
in get_iterable(vnfd
, 'vdu'):
2128 descriptor_config
= vdud
.get('vdu-configuration')
2129 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
2130 if vdur
.get("additionalParams"):
2131 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
2133 deploy_params_vdu
= deploy_params
2134 deploy_params_vdu
["OSM"] = self
._get
_osm
_params
(db_vnfr
, vdu_id
, vdu_count_index
=0)
2135 if descriptor_config
:
2138 for vdu_index
in range(int(vdud
.get("count", 1))):
2139 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2141 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2142 member_vnf_index
, vdu_id
, vdu_index
),
2145 nslcmop_id
=nslcmop_id
,
2151 member_vnf_index
=member_vnf_index
,
2152 vdu_index
=vdu_index
,
2154 deploy_params
=deploy_params_vdu
,
2155 descriptor_config
=descriptor_config
,
2156 base_folder
=base_folder
,
2157 task_instantiation_info
=tasks_dict_info
,
2160 for kdud
in get_iterable(vnfd
, 'kdu'):
2161 kdu_name
= kdud
["name"]
2162 descriptor_config
= kdud
.get('kdu-configuration')
2163 if descriptor_config
:
2167 kdur
= next(x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
)
2168 deploy_params_kdu
= {"OSM": self
._get
_osm
_params
(db_vnfr
)}
2169 if kdur
.get("additionalParams"):
2170 deploy_params_kdu
= self
._format
_additional
_params
(kdur
["additionalParams"])
2173 logging_text
=logging_text
,
2176 nslcmop_id
=nslcmop_id
,
2182 member_vnf_index
=member_vnf_index
,
2183 vdu_index
=vdu_index
,
2185 deploy_params
=deploy_params_kdu
,
2186 descriptor_config
=descriptor_config
,
2187 base_folder
=base_folder
,
2188 task_instantiation_info
=tasks_dict_info
,
2192 # Check if this NS has a charm configuration
2193 descriptor_config
= nsd
.get("ns-configuration")
2194 if descriptor_config
and descriptor_config
.get("juju"):
2197 member_vnf_index
= None
2203 # Get additional parameters
2204 deploy_params
= {"OSM": self
._get
_osm
_params
(db_vnfr
)}
2205 if db_nsr
.get("additionalParamsForNs"):
2206 deploy_params
.update(self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy()))
2207 base_folder
= nsd
["_admin"]["storage"]
2209 logging_text
=logging_text
,
2212 nslcmop_id
=nslcmop_id
,
2218 member_vnf_index
=member_vnf_index
,
2219 vdu_index
=vdu_index
,
2221 deploy_params
=deploy_params
,
2222 descriptor_config
=descriptor_config
,
2223 base_folder
=base_folder
,
2224 task_instantiation_info
=tasks_dict_info
,
2228 # rest of staff will be done at finally
2230 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2231 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
2233 except asyncio
.CancelledError
:
2234 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2235 exc
= "Operation was cancelled"
2236 except Exception as e
:
2237 exc
= traceback
.format_exc()
2238 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2241 error_list
.append(str(exc
))
2243 # wait for pending tasks
2245 stage
[1] = "Waiting for instantiate pending tasks."
2246 self
.logger
.debug(logging_text
+ stage
[1])
2247 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
2248 stage
, nslcmop_id
, nsr_id
=nsr_id
)
2249 stage
[1] = stage
[2] = ""
2250 except asyncio
.CancelledError
:
2251 error_list
.append("Cancelled")
2252 # TODO cancel all tasks
2253 except Exception as exc
:
2254 error_list
.append(str(exc
))
2256 # update operation-status
2257 db_nsr_update
["operational-status"] = "running"
2258 # let's begin with VCA 'configured' status (later we can change it)
2259 db_nsr_update
["config-status"] = "configured"
2260 for task
, task_name
in tasks_dict_info
.items():
2261 if not task
.done() or task
.cancelled() or task
.exception():
2262 if task_name
.startswith(self
.task_name_deploy_vca
):
2263 # A N2VC task is pending
2264 db_nsr_update
["config-status"] = "failed"
2266 # RO or KDU task is pending
2267 db_nsr_update
["operational-status"] = "failed"
2269 # update status at database
2271 error_detail
= ". ".join(error_list
)
2272 self
.logger
.error(logging_text
+ error_detail
)
2273 error_description_nslcmop
= '{} Detail: {}'.format(stage
[0], error_detail
)
2274 error_description_nsr
= 'Operation: INSTANTIATING.{}, {}'.format(nslcmop_id
, stage
[0])
2276 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2277 db_nslcmop_update
["detailed-status"] = error_detail
2278 nslcmop_operation_state
= "FAILED"
2282 error_description_nsr
= error_description_nslcmop
= None
2284 db_nsr_update
["detailed-status"] = "Done"
2285 db_nslcmop_update
["detailed-status"] = "Done"
2286 nslcmop_operation_state
= "COMPLETED"
2289 self
._write
_ns
_status
(
2292 current_operation
="IDLE",
2293 current_operation_id
=None,
2294 error_description
=error_description_nsr
,
2295 error_detail
=error_detail
,
2296 other_update
=db_nsr_update
2298 self
._write
_op
_status
(
2301 error_message
=error_description_nslcmop
,
2302 operation_state
=nslcmop_operation_state
,
2303 other_update
=db_nslcmop_update
,
2306 if nslcmop_operation_state
:
2308 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2309 "operationState": nslcmop_operation_state
},
2311 except Exception as e
:
2312 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2314 self
.logger
.debug(logging_text
+ "Exit")
2315 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2317 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int,
2318 timeout
: int = 3600, vca_type
: str = None) -> bool:
2321 # 1. find all relations for this VCA
2322 # 2. wait for other peers related
2326 vca_type
= vca_type
or "lxc_proxy_charm"
2328 # STEP 1: find all relations for this VCA
2331 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2332 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2335 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
2337 # read all ns-configuration relations
2338 ns_relations
= list()
2339 db_ns_relations
= deep_get(nsd
, ('ns-configuration', 'relation'))
2341 for r
in db_ns_relations
:
2342 # check if this VCA is in the relation
2343 if my_vca
.get('member-vnf-index') in\
2344 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2345 ns_relations
.append(r
)
2347 # read all vnf-configuration relations
2348 vnf_relations
= list()
2349 db_vnfd_list
= db_nsr
.get('vnfd-id')
2351 for vnfd
in db_vnfd_list
:
2352 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2353 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
2354 if db_vnf_relations
:
2355 for r
in db_vnf_relations
:
2356 # check if this VCA is in the relation
2357 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2358 vnf_relations
.append(r
)
2360 # if no relations, terminate
2361 if not ns_relations
and not vnf_relations
:
2362 self
.logger
.debug(logging_text
+ ' No relations')
2365 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
2372 if now
- start
>= timeout
:
2373 self
.logger
.error(logging_text
+ ' : timeout adding relations')
2376 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2377 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2379 # for each defined NS relation, find the VCA's related
2380 for r
in ns_relations
:
2381 from_vca_ee_id
= None
2383 from_vca_endpoint
= None
2384 to_vca_endpoint
= None
2385 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2386 for vca
in vca_list
:
2387 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
2388 and vca
.get('config_sw_installed'):
2389 from_vca_ee_id
= vca
.get('ee_id')
2390 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2391 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2392 and vca
.get('config_sw_installed'):
2393 to_vca_ee_id
= vca
.get('ee_id')
2394 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2395 if from_vca_ee_id
and to_vca_ee_id
:
2397 await self
.vca_map
[vca_type
].add_relation(
2398 ee_id_1
=from_vca_ee_id
,
2399 ee_id_2
=to_vca_ee_id
,
2400 endpoint_1
=from_vca_endpoint
,
2401 endpoint_2
=to_vca_endpoint
)
2402 # remove entry from relations list
2403 ns_relations
.remove(r
)
2405 # check failed peers
2407 vca_status_list
= db_nsr
.get('configurationStatus')
2409 for i
in range(len(vca_list
)):
2411 vca_status
= vca_status_list
[i
]
2412 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2413 if vca_status
.get('status') == 'BROKEN':
2414 # peer broken: remove relation from list
2415 ns_relations
.remove(r
)
2416 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2417 if vca_status
.get('status') == 'BROKEN':
2418 # peer broken: remove relation from list
2419 ns_relations
.remove(r
)
2424 # for each defined VNF relation, find the VCA's related
2425 for r
in vnf_relations
:
2426 from_vca_ee_id
= None
2428 from_vca_endpoint
= None
2429 to_vca_endpoint
= None
2430 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2431 for vca
in vca_list
:
2432 key_to_check
= "vdu_id"
2433 if vca
.get("vdu_id") is None:
2434 key_to_check
= "vnfd_id"
2435 if vca
.get(key_to_check
) == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2436 from_vca_ee_id
= vca
.get('ee_id')
2437 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2438 if vca
.get(key_to_check
) == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2439 to_vca_ee_id
= vca
.get('ee_id')
2440 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2441 if from_vca_ee_id
and to_vca_ee_id
:
2443 await self
.vca_map
[vca_type
].add_relation(
2444 ee_id_1
=from_vca_ee_id
,
2445 ee_id_2
=to_vca_ee_id
,
2446 endpoint_1
=from_vca_endpoint
,
2447 endpoint_2
=to_vca_endpoint
)
2448 # remove entry from relations list
2449 vnf_relations
.remove(r
)
2451 # check failed peers
2453 vca_status_list
= db_nsr
.get('configurationStatus')
2455 for i
in range(len(vca_list
)):
2457 vca_status
= vca_status_list
[i
]
2458 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2459 if vca_status
.get('status') == 'BROKEN':
2460 # peer broken: remove relation from list
2461 ns_relations
.remove(r
)
2462 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2463 if vca_status
.get('status') == 'BROKEN':
2464 # peer broken: remove relation from list
2465 ns_relations
.remove(r
)
2471 await asyncio
.sleep(5.0)
2473 if not ns_relations
and not vnf_relations
:
2474 self
.logger
.debug('Relations added')
2479 except Exception as e
:
2480 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2483 async def _install_kdu(self
, nsr_id
: str, nsr_db_path
: str, vnfr_data
: dict, kdu_index
: int, kdud
: dict,
2484 vnfd
: dict, k8s_instance_info
: dict, k8params
: dict = None, timeout
: int = 600):
2487 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
2489 db_dict_install
= {"collection": "nsrs",
2490 "filter": {"_id": nsr_id
},
2491 "path": nsr_db_path
}
2493 kdu_instance
= await self
.k8scluster_map
[k8sclustertype
].install(
2494 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2495 kdu_model
=k8s_instance_info
["kdu-model"],
2498 db_dict
=db_dict_install
,
2500 kdu_name
=k8s_instance_info
["kdu-name"],
2501 namespace
=k8s_instance_info
["namespace"])
2502 self
.update_db_2("nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
})
2504 # Obtain services to obtain management service ip
2505 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
2506 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2507 kdu_instance
=kdu_instance
,
2508 namespace
=k8s_instance_info
["namespace"])
2510 # Obtain management service info (if exists)
2511 vnfr_update_dict
= {}
2513 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
2514 mgmt_services
= [service
for service
in kdud
.get("service", []) if service
.get("mgmt-service")]
2515 for mgmt_service
in mgmt_services
:
2516 for service
in services
:
2517 if service
["name"].startswith(mgmt_service
["name"]):
2518 # Mgmt service found, Obtain service ip
2519 ip
= service
.get("external_ip", service
.get("cluster_ip"))
2520 if isinstance(ip
, list) and len(ip
) == 1:
2523 vnfr_update_dict
["kdur.{}.ip-address".format(kdu_index
)] = ip
2525 # Check if must update also mgmt ip at the vnf
2526 service_external_cp
= mgmt_service
.get("external-connection-point-ref")
2527 if service_external_cp
:
2528 if deep_get(vnfd
, ("mgmt-interface", "cp")) == service_external_cp
:
2529 vnfr_update_dict
["ip-address"] = ip
2533 self
.logger
.warn("Mgmt service name: {} not found".format(mgmt_service
["name"]))
2535 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
2536 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
2538 kdu_config
= kdud
.get("kdu-configuration")
2539 if kdu_config
and kdu_config
.get("initial-config-primitive") and kdu_config
.get("juju") is None:
2540 initial_config_primitive_list
= kdu_config
.get("initial-config-primitive")
2541 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
2543 for initial_config_primitive
in initial_config_primitive_list
:
2544 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, {})
2546 await asyncio
.wait_for(
2547 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
2548 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2549 kdu_instance
=kdu_instance
,
2550 primitive_name
=initial_config_primitive
["name"],
2551 params
=primitive_params_
, db_dict
={}),
2554 except Exception as e
:
2555 # Prepare update db with error and raise exception
2557 self
.update_db_2("nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)})
2558 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), {"kdur.{}.status".format(kdu_index
): "ERROR"})
2560 # ignore to keep original exception
2562 # reraise original error
2567 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2568 # Launch kdus if present in the descriptor
2570 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2572 async def _get_cluster_id(cluster_id
, cluster_type
):
2573 nonlocal k8scluster_id_2_uuic
2574 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2575 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2577 # check if K8scluster is creating and wait look if previous tasks in process
2578 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("k8scluster", cluster_id
)
2580 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(task_name
, cluster_id
)
2581 self
.logger
.debug(logging_text
+ text
)
2582 await asyncio
.wait(task_dependency
, timeout
=3600)
2584 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2585 if not db_k8scluster
:
2586 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2588 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2590 raise LcmException("K8s cluster '{}' has not been initialized for '{}'".format(cluster_id
,
2592 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2595 logging_text
+= "Deploy kdus: "
2598 db_nsr_update
= {"_admin.deployed.K8s": []}
2599 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2602 updated_cluster_list
= []
2604 for vnfr_data
in db_vnfrs
.values():
2605 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
2606 # Step 0: Prepare and set parameters
2607 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2608 vnfd_id
= vnfr_data
.get('vnfd-id')
2609 kdud
= next(kdud
for kdud
in db_vnfds
[vnfd_id
]["kdu"] if kdud
["name"] == kdur
["kdu-name"])
2610 namespace
= kdur
.get("k8s-namespace")
2611 if kdur
.get("helm-chart"):
2612 kdumodel
= kdur
["helm-chart"]
2613 k8sclustertype
= "helm-chart"
2614 elif kdur
.get("juju-bundle"):
2615 kdumodel
= kdur
["juju-bundle"]
2616 k8sclustertype
= "juju-bundle"
2618 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2619 "juju-bundle. Maybe an old NBI version is running".
2620 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2621 # check if kdumodel is a file and exists
2623 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2624 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2625 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2626 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["pkg-dir"], k8sclustertype
,
2628 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2629 kdumodel
= self
.fs
.path
+ filename
2630 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2632 except Exception: # it is not a file
2635 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2636 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2637 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2640 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2641 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2642 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2643 if del_repo_list
or added_repo_dict
:
2644 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2645 updated
= {'_admin.helm_charts_added.' +
2646 item
: name
for item
, name
in added_repo_dict
.items()}
2647 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2648 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2650 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2651 updated_cluster_list
.append(cluster_uuid
)
2654 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2655 kdur
["kdu-name"], k8s_cluster_id
)
2656 k8s_instance_info
= {"kdu-instance": None,
2657 "k8scluster-uuid": cluster_uuid
,
2658 "k8scluster-type": k8sclustertype
,
2659 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2660 "kdu-name": kdur
["kdu-name"],
2661 "kdu-model": kdumodel
,
2662 "namespace": namespace
}
2663 db_path
= "_admin.deployed.K8s.{}".format(index
)
2664 db_nsr_update
[db_path
] = k8s_instance_info
2665 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2667 task
= asyncio
.ensure_future(
2668 self
._install
_kdu
(nsr_id
, db_path
, vnfr_data
, kdu_index
, kdud
, db_vnfds
[vnfd_id
],
2669 k8s_instance_info
, k8params
=desc_params
, timeout
=600))
2670 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2671 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2675 except (LcmException
, asyncio
.CancelledError
):
2677 except Exception as e
:
2678 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2679 if isinstance(e
, (N2VCException
, DbException
)):
2680 self
.logger
.error(logging_text
+ msg
)
2682 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2683 raise LcmException(msg
)
2686 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2688 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2689 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2690 base_folder
, task_instantiation_info
, stage
):
2691 # launch instantiate_N2VC in a asyncio task and register task object
2692 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2693 # if not found, create one entry and update database
2694 # fill db_nsr._admin.deployed.VCA.<index>
2696 self
.logger
.debug(logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
))
2697 if descriptor_config
.get("juju"): # There is one execution envioronment of type juju
2698 ee_list
= [descriptor_config
]
2699 elif descriptor_config
.get("execution-environment-list"):
2700 ee_list
= descriptor_config
.get("execution-environment-list")
2701 else: # other types as script are not supported
2704 for ee_item
in ee_list
:
2705 self
.logger
.debug(logging_text
+ "_deploy_n2vc ee_item juju={}, helm={}".format(ee_item
.get('juju'),
2706 ee_item
.get("helm-chart")))
2707 ee_descriptor_id
= ee_item
.get("id")
2708 if ee_item
.get("juju"):
2709 vca_name
= ee_item
['juju'].get('charm')
2710 vca_type
= "lxc_proxy_charm" if ee_item
['juju'].get('charm') is not None else "native_charm"
2711 if ee_item
['juju'].get('cloud') == "k8s":
2712 vca_type
= "k8s_proxy_charm"
2713 elif ee_item
['juju'].get('proxy') is False:
2714 vca_type
= "native_charm"
2715 elif ee_item
.get("helm-chart"):
2716 vca_name
= ee_item
['helm-chart']
2719 self
.logger
.debug(logging_text
+ "skipping non juju neither charm configuration")
2723 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2724 if not vca_deployed
:
2726 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2727 vca_deployed
.get("vdu_id") == vdu_id
and \
2728 vca_deployed
.get("kdu_name") == kdu_name
and \
2729 vca_deployed
.get("vdu_count_index", 0) == vdu_index
and \
2730 vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
:
2733 # not found, create one.
2734 target
= "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
2736 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
2738 target
+= "/kdu/{}".format(kdu_name
)
2740 "target_element": target
,
2741 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
2742 "member-vnf-index": member_vnf_index
,
2744 "kdu_name": kdu_name
,
2745 "vdu_count_index": vdu_index
,
2746 "operational-status": "init", # TODO revise
2747 "detailed-status": "", # TODO revise
2748 "step": "initial-deploy", # TODO revise
2750 "vdu_name": vdu_name
,
2752 "ee_descriptor_id": ee_descriptor_id
2756 # create VCA and configurationStatus in db
2758 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2759 "configurationStatus.{}".format(vca_index
): dict()
2761 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2763 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2766 task_n2vc
= asyncio
.ensure_future(
2767 self
.instantiate_N2VC(
2768 logging_text
=logging_text
,
2769 vca_index
=vca_index
,
2775 vdu_index
=vdu_index
,
2776 deploy_params
=deploy_params
,
2777 config_descriptor
=descriptor_config
,
2778 base_folder
=base_folder
,
2779 nslcmop_id
=nslcmop_id
,
2783 ee_config_descriptor
=ee_item
2786 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2787 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2788 member_vnf_index
or "", vdu_id
or "")
2791 def _get_terminate_config_primitive(primitive_list
, vca_deployed
):
2792 """ Get a sorted terminate config primitive list. In case ee_descriptor_id is present at vca_deployed,
2793 it get only those primitives for this execution envirom"""
2795 primitive_list
= primitive_list
or []
2796 # filter primitives by ee_descriptor_id
2797 ee_descriptor_id
= vca_deployed
.get("ee_descriptor_id")
2798 primitive_list
= [p
for p
in primitive_list
if p
.get("execution-environment-ref") == ee_descriptor_id
]
2801 primitive_list
.sort(key
=lambda val
: int(val
['seq']))
2803 return primitive_list
2806 def _create_nslcmop(nsr_id
, operation
, params
):
2808 Creates a ns-lcm-opp content to be stored at database.
2809 :param nsr_id: internal id of the instance
2810 :param operation: instantiate, terminate, scale, action, ...
2811 :param params: user parameters for the operation
2812 :return: dictionary following SOL005 format
2814 # Raise exception if invalid arguments
2815 if not (nsr_id
and operation
and params
):
2817 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2823 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2824 "operationState": "PROCESSING",
2825 "statusEnteredTime": now
,
2826 "nsInstanceId": nsr_id
,
2827 "lcmOperationType": operation
,
2829 "isAutomaticInvocation": False,
2830 "operationParams": params
,
2831 "isCancelPending": False,
2833 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2834 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2839 def _format_additional_params(self
, params
):
2840 params
= params
or {}
2841 for key
, value
in params
.items():
2842 if str(value
).startswith("!!yaml "):
2843 params
[key
] = yaml
.safe_load(value
[7:])
2846 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2847 primitive
= seq
.get('name')
2848 primitive_params
= {}
2850 "member_vnf_index": vnf_index
,
2851 "primitive": primitive
,
2852 "primitive_params": primitive_params
,
2855 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2859 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2860 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2861 if op
.get('operationState') == 'COMPLETED':
2862 # b. Skip sub-operation
2863 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2864 return self
.SUBOPERATION_STATUS_SKIP
2866 # c. retry executing sub-operation
2867 # The sub-operation exists, and operationState != 'COMPLETED'
2868 # Update operationState = 'PROCESSING' to indicate a retry.
2869 operationState
= 'PROCESSING'
2870 detailed_status
= 'In progress'
2871 self
._update
_suboperation
_status
(
2872 db_nslcmop
, op_index
, operationState
, detailed_status
)
2873 # Return the sub-operation index
2874 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2875 # with arguments extracted from the sub-operation
2878 # Find a sub-operation where all keys in a matching dictionary must match
2879 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2880 def _find_suboperation(self
, db_nslcmop
, match
):
2881 if db_nslcmop
and match
:
2882 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2883 for i
, op
in enumerate(op_list
):
2884 if all(op
.get(k
) == match
[k
] for k
in match
):
2886 return self
.SUBOPERATION_STATUS_NOT_FOUND
2888 # Update status for a sub-operation given its index
2889 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2890 # Update DB for HA tasks
2891 q_filter
= {'_id': db_nslcmop
['_id']}
2892 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2893 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2894 self
.db
.set_one("nslcmops",
2896 update_dict
=update_dict
,
2897 fail_on_empty
=False)
2899 # Add sub-operation, return the index of the added sub-operation
2900 # Optionally, set operationState, detailed-status, and operationType
2901 # Status and type are currently set for 'scale' sub-operations:
2902 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2903 # 'detailed-status' : status message
2904 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2905 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2906 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2907 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2908 RO_nsr_id
=None, RO_scaling_info
=None):
2910 return self
.SUBOPERATION_STATUS_NOT_FOUND
2911 # Get the "_admin.operations" list, if it exists
2912 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2913 op_list
= db_nslcmop_admin
.get('operations')
2914 # Create or append to the "_admin.operations" list
2915 new_op
= {'member_vnf_index': vnf_index
,
2917 'vdu_count_index': vdu_count_index
,
2918 'primitive': primitive
,
2919 'primitive_params': mapped_primitive_params
}
2921 new_op
['operationState'] = operationState
2923 new_op
['detailed-status'] = detailed_status
2925 new_op
['lcmOperationType'] = operationType
2927 new_op
['RO_nsr_id'] = RO_nsr_id
2929 new_op
['RO_scaling_info'] = RO_scaling_info
2931 # No existing operations, create key 'operations' with current operation as first list element
2932 db_nslcmop_admin
.update({'operations': [new_op
]})
2933 op_list
= db_nslcmop_admin
.get('operations')
2935 # Existing operations, append operation to list
2936 op_list
.append(new_op
)
2938 db_nslcmop_update
= {'_admin.operations': op_list
}
2939 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2940 op_index
= len(op_list
) - 1
2943 # Helper methods for scale() sub-operations
2945 # pre-scale/post-scale:
2946 # Check for 3 different cases:
2947 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2948 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2949 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2950 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2951 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2952 # Find this sub-operation
2953 if RO_nsr_id
and RO_scaling_info
:
2954 operationType
= 'SCALE-RO'
2956 'member_vnf_index': vnf_index
,
2957 'RO_nsr_id': RO_nsr_id
,
2958 'RO_scaling_info': RO_scaling_info
,
2962 'member_vnf_index': vnf_index
,
2963 'primitive': vnf_config_primitive
,
2964 'primitive_params': primitive_params
,
2965 'lcmOperationType': operationType
2967 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2968 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2969 # a. New sub-operation
2970 # The sub-operation does not exist, add it.
2971 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2972 # The following parameters are set to None for all kind of scaling:
2974 vdu_count_index
= None
2976 if RO_nsr_id
and RO_scaling_info
:
2977 vnf_config_primitive
= None
2978 primitive_params
= None
2981 RO_scaling_info
= None
2982 # Initial status for sub-operation
2983 operationState
= 'PROCESSING'
2984 detailed_status
= 'In progress'
2985 # Add sub-operation for pre/post-scaling (zero or more operations)
2986 self
._add
_suboperation
(db_nslcmop
,
2991 vnf_config_primitive
,
2998 return self
.SUBOPERATION_STATUS_NEW
3000 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
3001 # or op_index (operationState != 'COMPLETED')
3002 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
3004 # Function to return execution_environment id
3006 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
3007 # TODO vdu_index_count
3008 for vca
in vca_deployed_list
:
3009 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
3012 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
,
3013 vca_index
, destroy_ee
=True, exec_primitives
=True):
3015 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
3016 :param logging_text:
3018 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
3019 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
3020 :param vca_index: index in the database _admin.deployed.VCA
3021 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
3022 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
3023 not executed properly
3024 :return: None or exception
3028 logging_text
+ " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
3029 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
3033 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
3035 # execute terminate_primitives
3037 terminate_primitives
= self
._get
_terminate
_config
_primitive
(
3038 config_descriptor
.get("terminate-config-primitive"), vca_deployed
)
3039 vdu_id
= vca_deployed
.get("vdu_id")
3040 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3041 vdu_name
= vca_deployed
.get("vdu_name")
3042 vnf_index
= vca_deployed
.get("member-vnf-index")
3043 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3044 for seq
in terminate_primitives
:
3045 # For each sequence in list, get primitive and call _ns_execute_primitive()
3046 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3047 vnf_index
, seq
.get("name"))
3048 self
.logger
.debug(logging_text
+ step
)
3049 # Create the primitive for each sequence, i.e. "primitive": "touch"
3050 primitive
= seq
.get('name')
3051 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
3054 self
._add
_suboperation
(db_nslcmop
,
3060 mapped_primitive_params
)
3061 # Sub-operations: Call _ns_execute_primitive() instead of action()
3063 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
3064 mapped_primitive_params
,
3066 except LcmException
:
3067 # this happens when VCA is not deployed. In this case it is not needed to terminate
3069 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
3070 if result
not in result_ok
:
3071 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
3072 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
3073 # set that this VCA do not need terminated
3074 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
3075 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
3077 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3078 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3081 await self
.vca_map
[vca_type
].delete_execution_environment(vca_deployed
["ee_id"])
3083 async def _delete_all_N2VC(self
, db_nsr
: dict):
3084 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
3085 namespace
= "." + db_nsr
["_id"]
3087 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
3088 except N2VCNotFound
: # already deleted. Skip
3090 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
3092 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
3094 Terminates a deployment from RO
3095 :param logging_text:
3096 :param nsr_deployed: db_nsr._admin.deployed
3099 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3100 this method will update only the index 2, but it will write on database the concatenated content of the list
3105 ro_nsr_id
= ro_delete_action
= None
3106 if nsr_deployed
and nsr_deployed
.get("RO"):
3107 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3108 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3111 stage
[2] = "Deleting ns from VIM."
3112 db_nsr_update
["detailed-status"] = " ".join(stage
)
3113 self
._write
_op
_status
(nslcmop_id
, stage
)
3114 self
.logger
.debug(logging_text
+ stage
[2])
3115 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3116 self
._write
_op
_status
(nslcmop_id
, stage
)
3117 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3118 ro_delete_action
= desc
["action_id"]
3119 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
3120 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3121 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3122 if ro_delete_action
:
3123 # wait until NS is deleted from VIM
3124 stage
[2] = "Waiting ns deleted from VIM."
3125 detailed_status_old
= None
3126 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
3128 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3129 self
._write
_op
_status
(nslcmop_id
, stage
)
3131 delete_timeout
= 20 * 60 # 20 minutes
3132 while delete_timeout
> 0:
3133 desc
= await self
.RO
.show(
3135 item_id_name
=ro_nsr_id
,
3136 extra_item
="action",
3137 extra_item_id
=ro_delete_action
)
3140 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3142 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3143 if ns_status
== "ERROR":
3144 raise ROclient
.ROClientException(ns_status_info
)
3145 elif ns_status
== "BUILD":
3146 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3147 elif ns_status
== "ACTIVE":
3148 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3149 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3152 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3153 if stage
[2] != detailed_status_old
:
3154 detailed_status_old
= stage
[2]
3155 db_nsr_update
["detailed-status"] = " ".join(stage
)
3156 self
._write
_op
_status
(nslcmop_id
, stage
)
3157 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3158 await asyncio
.sleep(5, loop
=self
.loop
)
3160 else: # delete_timeout <= 0:
3161 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
3163 except Exception as e
:
3164 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3165 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3166 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3167 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3168 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3169 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
3170 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3171 failed_detail
.append("delete conflict: {}".format(e
))
3172 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
3174 failed_detail
.append("delete error: {}".format(e
))
3175 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
3178 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
3179 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
3181 stage
[2] = "Deleting nsd from RO."
3182 db_nsr_update
["detailed-status"] = " ".join(stage
)
3183 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3184 self
._write
_op
_status
(nslcmop_id
, stage
)
3185 await self
.RO
.delete("nsd", ro_nsd_id
)
3186 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
3187 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3188 except Exception as e
:
3189 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3190 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3191 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
3192 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3193 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
3194 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3196 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
3197 self
.logger
.error(logging_text
+ failed_detail
[-1])
3199 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
3200 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
3201 if not vnf_deployed
or not vnf_deployed
["id"]:
3204 ro_vnfd_id
= vnf_deployed
["id"]
3205 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
3206 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
3207 db_nsr_update
["detailed-status"] = " ".join(stage
)
3208 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3209 self
._write
_op
_status
(nslcmop_id
, stage
)
3210 await self
.RO
.delete("vnfd", ro_vnfd_id
)
3211 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
3212 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3213 except Exception as e
:
3214 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3215 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3216 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
3217 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3218 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
3219 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3221 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
3222 self
.logger
.error(logging_text
+ failed_detail
[-1])
3225 stage
[2] = "Error deleting from VIM"
3227 stage
[2] = "Deleted from VIM"
3228 db_nsr_update
["detailed-status"] = " ".join(stage
)
3229 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3230 self
._write
_op
_status
(nslcmop_id
, stage
)
3233 raise LcmException("; ".join(failed_detail
))
3235 async def terminate(self
, nsr_id
, nslcmop_id
):
3236 # Try to lock HA task here
3237 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3238 if not task_is_locked_by_me
:
3241 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
3242 self
.logger
.debug(logging_text
+ "Enter")
3243 timeout_ns_terminate
= self
.timeout_ns_terminate
3246 operation_params
= None
3248 error_list
= [] # annotates all failed error messages
3249 db_nslcmop_update
= {}
3250 autoremove
= False # autoremove after terminated
3251 tasks_dict_info
= {}
3253 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
3254 # ^ contains [stage, step, VIM-status]
3256 # wait for any previous tasks in process
3257 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
3259 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
3260 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3261 operation_params
= db_nslcmop
.get("operationParams") or {}
3262 if operation_params
.get("timeout_ns_terminate"):
3263 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
3264 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
3265 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3267 db_nsr_update
["operational-status"] = "terminating"
3268 db_nsr_update
["config-status"] = "terminating"
3269 self
._write
_ns
_status
(
3271 ns_state
="TERMINATING",
3272 current_operation
="TERMINATING",
3273 current_operation_id
=nslcmop_id
,
3274 other_update
=db_nsr_update
3276 self
._write
_op
_status
(
3281 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
3282 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
3285 stage
[1] = "Getting vnf descriptors from db."
3286 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
3287 db_vnfds_from_id
= {}
3288 db_vnfds_from_member_index
= {}
3290 for vnfr
in db_vnfrs_list
:
3291 vnfd_id
= vnfr
["vnfd-id"]
3292 if vnfd_id
not in db_vnfds_from_id
:
3293 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
3294 db_vnfds_from_id
[vnfd_id
] = vnfd
3295 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
3297 # Destroy individual execution environments when there are terminating primitives.
3298 # Rest of EE will be deleted at once
3299 # TODO - check before calling _destroy_N2VC
3300 # if not operation_params.get("skip_terminate_primitives"):#
3301 # or not vca.get("needed_terminate"):
3302 stage
[0] = "Stage 2/3 execute terminating primitives."
3303 self
.logger
.debug(logging_text
+ stage
[0])
3304 stage
[1] = "Looking execution environment that needs terminate."
3305 self
.logger
.debug(logging_text
+ stage
[1])
3306 # self.logger.debug("nsr_deployed: {}".format(nsr_deployed))
3307 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
3308 config_descriptor
= None
3309 if not vca
or not vca
.get("ee_id"):
3311 if not vca
.get("member-vnf-index"):
3313 config_descriptor
= db_nsr
.get("ns-configuration")
3314 elif vca
.get("vdu_id"):
3315 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3316 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
3318 config_descriptor
= vdud
.get("vdu-configuration")
3319 elif vca
.get("kdu_name"):
3320 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3321 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
3323 config_descriptor
= kdud
.get("kdu-configuration")
3325 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
3326 vca_type
= vca
.get("type")
3327 exec_terminate_primitives
= (not operation_params
.get("skip_terminate_primitives") and
3328 vca
.get("needed_terminate"))
3329 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
3330 # pending native charms
3331 destroy_ee
= True if vca_type
in ("helm", "native_charm") else False
3332 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
3333 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
3334 task
= asyncio
.ensure_future(
3335 self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
, vca_index
,
3336 destroy_ee
, exec_terminate_primitives
))
3337 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
3339 # wait for pending tasks of terminate primitives
3341 self
.logger
.debug(logging_text
+ 'Waiting for tasks {}'.format(list(tasks_dict_info
.keys())))
3342 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
3343 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
3345 tasks_dict_info
.clear()
3347 return # raise LcmException("; ".join(error_list))
3349 # remove All execution environments at once
3350 stage
[0] = "Stage 3/3 delete all."
3352 if nsr_deployed
.get("VCA"):
3353 stage
[1] = "Deleting all execution environments."
3354 self
.logger
.debug(logging_text
+ stage
[1])
3355 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
3356 timeout
=self
.timeout_charm_delete
))
3357 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
3358 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
3360 # Delete from k8scluster
3361 stage
[1] = "Deleting KDUs."
3362 self
.logger
.debug(logging_text
+ stage
[1])
3363 # print(nsr_deployed)
3364 for kdu
in get_iterable(nsr_deployed
, "K8s"):
3365 if not kdu
or not kdu
.get("kdu-instance"):
3367 kdu_instance
= kdu
.get("kdu-instance")
3368 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
3369 task_delete_kdu_instance
= asyncio
.ensure_future(
3370 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
3371 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3372 kdu_instance
=kdu_instance
))
3374 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
3375 format(kdu
.get("k8scluster-type")))
3377 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
3380 stage
[1] = "Deleting ns from VIM."
3382 task_delete_ro
= asyncio
.ensure_future(
3383 self
._terminate
_ng
_ro
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3385 task_delete_ro
= asyncio
.ensure_future(
3386 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3387 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
3389 # rest of staff will be done at finally
3391 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
3392 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3394 except asyncio
.CancelledError
:
3395 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
3396 exc
= "Operation was cancelled"
3397 except Exception as e
:
3398 exc
= traceback
.format_exc()
3399 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
3402 error_list
.append(str(exc
))
3404 # wait for pending tasks
3406 stage
[1] = "Waiting for terminate pending tasks."
3407 self
.logger
.debug(logging_text
+ stage
[1])
3408 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
3410 stage
[1] = stage
[2] = ""
3411 except asyncio
.CancelledError
:
3412 error_list
.append("Cancelled")
3413 # TODO cancell all tasks
3414 except Exception as exc
:
3415 error_list
.append(str(exc
))
3416 # update status at database
3418 error_detail
= "; ".join(error_list
)
3419 # self.logger.error(logging_text + error_detail)
3420 error_description_nslcmop
= '{} Detail: {}'.format(stage
[0], error_detail
)
3421 error_description_nsr
= 'Operation: TERMINATING.{}, {}.'.format(nslcmop_id
, stage
[0])
3423 db_nsr_update
["operational-status"] = "failed"
3424 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
3425 db_nslcmop_update
["detailed-status"] = error_detail
3426 nslcmop_operation_state
= "FAILED"
3430 error_description_nsr
= error_description_nslcmop
= None
3431 ns_state
= "NOT_INSTANTIATED"
3432 db_nsr_update
["operational-status"] = "terminated"
3433 db_nsr_update
["detailed-status"] = "Done"
3434 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
3435 db_nslcmop_update
["detailed-status"] = "Done"
3436 nslcmop_operation_state
= "COMPLETED"
3439 self
._write
_ns
_status
(
3442 current_operation
="IDLE",
3443 current_operation_id
=None,
3444 error_description
=error_description_nsr
,
3445 error_detail
=error_detail
,
3446 other_update
=db_nsr_update
3448 self
._write
_op
_status
(
3451 error_message
=error_description_nslcmop
,
3452 operation_state
=nslcmop_operation_state
,
3453 other_update
=db_nslcmop_update
,
3455 if ns_state
== "NOT_INSTANTIATED":
3457 self
.db
.set_list("vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "NOT_INSTANTIATED"})
3458 except DbException
as e
:
3459 self
.logger
.warn(logging_text
+ 'Error writing VNFR status for nsr-id-ref: {} -> {}'.
3461 if operation_params
:
3462 autoremove
= operation_params
.get("autoremove", False)
3463 if nslcmop_operation_state
:
3465 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3466 "operationState": nslcmop_operation_state
,
3467 "autoremove": autoremove
},
3469 except Exception as e
:
3470 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3472 self
.logger
.debug(logging_text
+ "Exit")
3473 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
3475 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
3477 error_detail_list
= []
3479 pending_tasks
= list(created_tasks_info
.keys())
3480 num_tasks
= len(pending_tasks
)
3482 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3483 self
._write
_op
_status
(nslcmop_id
, stage
)
3484 while pending_tasks
:
3486 _timeout
= timeout
+ time_start
- time()
3487 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
3488 return_when
=asyncio
.FIRST_COMPLETED
)
3489 num_done
+= len(done
)
3490 if not done
: # Timeout
3491 for task
in pending_tasks
:
3492 new_error
= created_tasks_info
[task
] + ": Timeout"
3493 error_detail_list
.append(new_error
)
3494 error_list
.append(new_error
)
3497 if task
.cancelled():
3500 exc
= task
.exception()
3502 if isinstance(exc
, asyncio
.TimeoutError
):
3504 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
3505 error_list
.append(created_tasks_info
[task
])
3506 error_detail_list
.append(new_error
)
3507 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
3509 self
.logger
.error(logging_text
+ new_error
)
3511 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
3512 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
3514 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
3515 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3517 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
3518 if nsr_id
: # update also nsr
3519 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
3520 "errorDetail": ". ".join(error_detail_list
)})
3521 self
._write
_op
_status
(nslcmop_id
, stage
)
3522 return error_detail_list
3524 def _map_primitive_params(self
, primitive_desc
, params
, instantiation_params
):
3526 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
3527 The default-value is used. If it is between < > it look for a value at instantiation_params
3528 :param primitive_desc: portion of VNFD/NSD that describes primitive
3529 :param params: Params provided by user
3530 :param instantiation_params: Instantiation params provided by user
3531 :return: a dictionary with the calculated params
3533 calculated_params
= {}
3534 for parameter
in primitive_desc
.get("parameter", ()):
3535 param_name
= parameter
["name"]
3536 if param_name
in params
:
3537 calculated_params
[param_name
] = params
[param_name
]
3538 elif "default-value" in parameter
or "value" in parameter
:
3539 if "value" in parameter
:
3540 calculated_params
[param_name
] = parameter
["value"]
3542 calculated_params
[param_name
] = parameter
["default-value"]
3543 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3544 and calculated_params
[param_name
].endswith(">"):
3545 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3546 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3548 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3549 format(calculated_params
[param_name
], primitive_desc
["name"]))
3551 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3552 format(param_name
, primitive_desc
["name"]))
3554 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3555 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3557 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3558 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3559 if parameter
.get("data-type") == "INTEGER":
3561 calculated_params
[param_name
] = int(calculated_params
[param_name
])
3562 except ValueError: # error converting string to int
3564 "Parameter {} of primitive {} must be integer".format(param_name
, primitive_desc
["name"]))
3565 elif parameter
.get("data-type") == "BOOLEAN":
3566 calculated_params
[param_name
] = not ((str(calculated_params
[param_name
])).lower() == 'false')
3568 # add always ns_config_info if primitive name is config
3569 if primitive_desc
["name"] == "config":
3570 if "ns_config_info" in instantiation_params
:
3571 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3572 calculated_params
["VCA"] = self
.vca_config
3573 return calculated_params
3575 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None,
3576 ee_descriptor_id
=None):
3577 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3578 for vca
in deployed_vca
:
3581 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3583 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3585 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3587 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
3591 # vca_deployed not found
3592 raise LcmException("charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
3593 " is not deployed".format(member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
,
3597 ee_id
= vca
.get("ee_id")
3598 vca_type
= vca
.get("type", "lxc_proxy_charm") # default value for backward compatibility - proxy charm
3600 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3601 "execution environment"
3602 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3603 return ee_id
, vca_type
3605 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3606 retries_interval
=30, timeout
=None,
3607 vca_type
=None, db_dict
=None) -> (str, str):
3609 if primitive
== "config":
3610 primitive_params
= {"params": primitive_params
}
3612 vca_type
= vca_type
or "lxc_proxy_charm"
3616 output
= await asyncio
.wait_for(
3617 self
.vca_map
[vca_type
].exec_primitive(
3619 primitive_name
=primitive
,
3620 params_dict
=primitive_params
,
3621 progress_timeout
=self
.timeout_progress_primitive
,
3622 total_timeout
=self
.timeout_primitive
,
3624 timeout
=timeout
or self
.timeout_primitive
)
3627 except asyncio
.CancelledError
:
3629 except Exception as e
: # asyncio.TimeoutError
3630 if isinstance(e
, asyncio
.TimeoutError
):
3634 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3636 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3638 return 'FAILED', str(e
)
3640 return 'COMPLETED', output
3642 except (LcmException
, asyncio
.CancelledError
):
3644 except Exception as e
:
3645 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3647 async def action(self
, nsr_id
, nslcmop_id
):
3649 # Try to lock HA task here
3650 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3651 if not task_is_locked_by_me
:
3654 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3655 self
.logger
.debug(logging_text
+ "Enter")
3656 # get all needed from database
3660 db_nslcmop_update
= {}
3661 nslcmop_operation_state
= None
3662 error_description_nslcmop
= None
3665 # wait for any previous tasks in process
3666 step
= "Waiting for previous operations to terminate"
3667 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3669 self
._write
_ns
_status
(
3672 current_operation
="RUNNING ACTION",
3673 current_operation_id
=nslcmop_id
3676 step
= "Getting information from database"
3677 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3678 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3680 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3681 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3682 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3683 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3684 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3685 primitive
= db_nslcmop
["operationParams"]["primitive"]
3686 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3687 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3690 step
= "Getting vnfr from database"
3691 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3692 step
= "Getting vnfd from database"
3693 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3695 step
= "Getting nsd from database"
3696 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3698 # for backward compatibility
3699 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3700 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3701 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3702 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3704 # look for primitive
3705 config_primitive_desc
= descriptor_configuration
= None
3707 for vdu
in get_iterable(db_vnfd
, "vdu"):
3708 if vdu_id
== vdu
["id"]:
3709 descriptor_configuration
= vdu
.get("vdu-configuration")
3712 for kdu
in get_iterable(db_vnfd
, "kdu"):
3713 if kdu_name
== kdu
["name"]:
3714 descriptor_configuration
= kdu
.get("kdu-configuration")
3717 descriptor_configuration
= db_vnfd
.get("vnf-configuration")
3719 descriptor_configuration
= db_nsd
.get("ns-configuration")
3721 if descriptor_configuration
and descriptor_configuration
.get("config-primitive"):
3722 for config_primitive
in descriptor_configuration
["config-primitive"]:
3723 if config_primitive
["name"] == primitive
:
3724 config_primitive_desc
= config_primitive
3727 if not config_primitive_desc
:
3728 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3729 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3731 primitive_name
= primitive
3732 ee_descriptor_id
= None
3734 primitive_name
= config_primitive_desc
.get("execution-environment-primitive", primitive
)
3735 ee_descriptor_id
= config_primitive_desc
.get("execution-environment-ref")
3739 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3740 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3742 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3743 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3745 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3747 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3750 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3752 # TODO check if ns is in a proper status
3753 if kdu_name
and (primitive_name
in ("upgrade", "rollback", "status") or kdu_action
):
3754 # kdur and desc_params already set from before
3755 if primitive_params
:
3756 desc_params
.update(primitive_params
)
3757 # TODO Check if we will need something at vnf level
3758 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3759 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3762 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3764 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3765 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3766 raise LcmException(msg
)
3768 db_dict
= {"collection": "nsrs",
3769 "filter": {"_id": nsr_id
},
3770 "path": "_admin.deployed.K8s.{}".format(index
)}
3771 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
))
3772 step
= "Executing kdu {}".format(primitive_name
)
3773 if primitive_name
== "upgrade":
3774 if desc_params
.get("kdu_model"):
3775 kdu_model
= desc_params
.get("kdu_model")
3776 del desc_params
["kdu_model"]
3778 kdu_model
= kdu
.get("kdu-model")
3779 parts
= kdu_model
.split(sep
=":")
3781 kdu_model
= parts
[0]
3783 detailed_status
= await asyncio
.wait_for(
3784 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3785 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3786 kdu_instance
=kdu
.get("kdu-instance"),
3787 atomic
=True, kdu_model
=kdu_model
,
3788 params
=desc_params
, db_dict
=db_dict
,
3789 timeout
=timeout_ns_action
),
3790 timeout
=timeout_ns_action
+ 10)
3791 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3792 elif primitive_name
== "rollback":
3793 detailed_status
= await asyncio
.wait_for(
3794 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3795 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3796 kdu_instance
=kdu
.get("kdu-instance"),
3798 timeout
=timeout_ns_action
)
3799 elif primitive_name
== "status":
3800 detailed_status
= await asyncio
.wait_for(
3801 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3802 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3803 kdu_instance
=kdu
.get("kdu-instance")),
3804 timeout
=timeout_ns_action
)
3806 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3807 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3809 detailed_status
= await asyncio
.wait_for(
3810 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3811 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3812 kdu_instance
=kdu_instance
,
3813 primitive_name
=primitive_name
,
3814 params
=params
, db_dict
=db_dict
,
3815 timeout
=timeout_ns_action
),
3816 timeout
=timeout_ns_action
)
3819 nslcmop_operation_state
= 'COMPLETED'
3821 detailed_status
= ''
3822 nslcmop_operation_state
= 'FAILED'
3824 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3825 member_vnf_index
=vnf_index
,
3827 vdu_count_index
=vdu_count_index
,
3828 ee_descriptor_id
=ee_descriptor_id
)
3829 db_nslcmop_notif
= {"collection": "nslcmops",
3830 "filter": {"_id": nslcmop_id
},
3831 "path": "admin.VCA"}
3832 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3834 primitive
=primitive_name
,
3835 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3836 timeout
=timeout_ns_action
,
3838 db_dict
=db_nslcmop_notif
)
3840 db_nslcmop_update
["detailed-status"] = detailed_status
3841 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3842 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3844 return # database update is called inside finally
3846 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3847 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3849 except asyncio
.CancelledError
:
3850 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3851 exc
= "Operation was cancelled"
3852 except asyncio
.TimeoutError
:
3853 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3855 except Exception as e
:
3856 exc
= traceback
.format_exc()
3857 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3860 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3861 "FAILED {}: {}".format(step
, exc
)
3862 nslcmop_operation_state
= "FAILED"
3864 self
._write
_ns
_status
(
3866 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3867 current_operation
="IDLE",
3868 current_operation_id
=None,
3869 # error_description=error_description_nsr,
3870 # error_detail=error_detail,
3871 other_update
=db_nsr_update
3874 self
._write
_op
_status
(
3877 error_message
=error_description_nslcmop
,
3878 operation_state
=nslcmop_operation_state
,
3879 other_update
=db_nslcmop_update
,
3882 if nslcmop_operation_state
:
3884 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3885 "operationState": nslcmop_operation_state
},
3887 except Exception as e
:
3888 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3889 self
.logger
.debug(logging_text
+ "Exit")
3890 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3891 return nslcmop_operation_state
, detailed_status
3893 async def scale(self
, nsr_id
, nslcmop_id
):
3895 # Try to lock HA task here
3896 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3897 if not task_is_locked_by_me
:
3900 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3901 self
.logger
.debug(logging_text
+ "Enter")
3902 # get all needed from database
3905 db_nslcmop_update
= {}
3906 nslcmop_operation_state
= None
3909 # in case of error, indicates what part of scale was failed to put nsr at error status
3910 scale_process
= None
3911 old_operational_status
= ""
3912 old_config_status
= ""
3914 # wait for any previous tasks in process
3915 step
= "Waiting for previous operations to terminate"
3916 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3918 self
._write
_ns
_status
(
3921 current_operation
="SCALING",
3922 current_operation_id
=nslcmop_id
3925 step
= "Getting nslcmop from database"
3926 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3927 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3928 step
= "Getting nsr from database"
3929 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3931 old_operational_status
= db_nsr
["operational-status"]
3932 old_config_status
= db_nsr
["config-status"]
3933 step
= "Parsing scaling parameters"
3934 # self.logger.debug(step)
3935 db_nsr_update
["operational-status"] = "scaling"
3936 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3937 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3940 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3941 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3942 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3943 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3944 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3947 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3948 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3949 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3950 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3951 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3953 # for backward compatibility
3954 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3955 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3956 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3957 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3959 step
= "Getting vnfr from database"
3960 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3961 step
= "Getting vnfd from database"
3962 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3964 step
= "Getting scaling-group-descriptor"
3965 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3966 if scaling_descriptor
["name"] == scaling_group
:
3969 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3970 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3973 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3974 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3975 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3978 # TODO check if ns is in a proper status
3979 step
= "Sending scale order to VIM"
3981 if not db_nsr
["_admin"].get("scaling-group"):
3982 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3983 admin_scale_index
= 0
3985 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3986 if admin_scale_info
["name"] == scaling_group
:
3987 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3989 else: # not found, set index one plus last element and add new entry with the name
3990 admin_scale_index
+= 1
3991 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3992 RO_scaling_info
= []
3993 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3994 if scaling_type
== "SCALE_OUT":
3995 # count if max-instance-count is reached
3996 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3997 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3998 if nb_scale_op
>= max_instance_count
:
3999 raise LcmException("reached the limit of {} (max-instance-count) "
4000 "scaling-out operations for the "
4001 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
4004 vdu_scaling_info
["scaling_direction"] = "OUT"
4005 vdu_scaling_info
["vdu-create"] = {}
4006 for vdu_scale_info
in scaling_descriptor
["vdu"]:
4007 vdud
= next(vdu
for vdu
in db_vnfd
.get("vdu") if vdu
["id"] == vdu_scale_info
["vdu-id-ref"])
4008 vdu_index
= len([x
for x
in db_vnfr
.get("vdur", ())
4009 if x
.get("vdu-id-ref") == vdu_scale_info
["vdu-id-ref"] and
4010 x
.get("member-vnf-index-ref") == vnf_index
])
4011 cloud_init_text
= self
._get
_cloud
_init
(vdud
, db_vnfd
)
4013 additional_params
= self
._get
_vdu
_additional
_params
(db_vnfr
, vdud
["id"]) or {}
4014 cloud_init_list
= []
4015 for x
in range(vdu_scale_info
.get("count", 1)):
4017 # TODO Information of its own ip is not available because db_vnfr is not updated.
4018 additional_params
["OSM"] = self
._get
_osm
_params
(db_vnfr
, vdu_scale_info
["vdu-id-ref"],
4020 cloud_init_list
.append(self
._parse
_cloud
_init
(cloud_init_text
, additional_params
,
4021 db_vnfd
["id"], vdud
["id"]))
4022 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
4023 "type": "create", "count": vdu_scale_info
.get("count", 1)})
4025 RO_scaling_info
[-1]["cloud_init"] = cloud_init_list
4026 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
4028 elif scaling_type
== "SCALE_IN":
4029 # count if min-instance-count is reached
4030 min_instance_count
= 0
4031 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
4032 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
4033 if nb_scale_op
<= min_instance_count
:
4034 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
4035 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
4037 vdu_scaling_info
["scaling_direction"] = "IN"
4038 vdu_scaling_info
["vdu-delete"] = {}
4039 for vdu_scale_info
in scaling_descriptor
["vdu"]:
4040 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
4041 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
4042 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
4044 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
4045 vdu_create
= vdu_scaling_info
.get("vdu-create")
4046 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
4047 if vdu_scaling_info
["scaling_direction"] == "IN":
4048 for vdur
in reversed(db_vnfr
["vdur"]):
4049 if vdu_delete
.get(vdur
["vdu-id-ref"]):
4050 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
4051 vdu_scaling_info
["vdu"].append({
4052 "name": vdur
["name"],
4053 "vdu_id": vdur
["vdu-id-ref"],
4056 for interface
in vdur
["interfaces"]:
4057 vdu_scaling_info
["vdu"][-1]["interface"].append({
4058 "name": interface
["name"],
4059 "ip_address": interface
["ip-address"],
4060 "mac_address": interface
.get("mac-address"),
4062 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
4065 step
= "Executing pre-scale vnf-config-primitive"
4066 if scaling_descriptor
.get("scaling-config-action"):
4067 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
4068 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
4069 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
4070 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
4071 step
= db_nslcmop_update
["detailed-status"] = \
4072 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
4074 # look for primitive
4075 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
4076 if config_primitive
["name"] == vnf_config_primitive
:
4080 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
4081 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
4082 "primitive".format(scaling_group
, vnf_config_primitive
))
4084 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
4085 if db_vnfr
.get("additionalParamsForVnf"):
4086 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
4088 scale_process
= "VCA"
4089 db_nsr_update
["config-status"] = "configuring pre-scaling"
4090 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
4092 # Pre-scale retry check: Check if this sub-operation has been executed before
4093 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4094 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
4095 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4096 # Skip sub-operation
4097 result
= 'COMPLETED'
4098 result_detail
= 'Done'
4099 self
.logger
.debug(logging_text
+
4100 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
4101 vnf_config_primitive
, result
, result_detail
))
4103 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4104 # New sub-operation: Get index of this sub-operation
4105 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4106 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
4107 format(vnf_config_primitive
))
4109 # retry: Get registered params for this existing sub-operation
4110 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4111 vnf_index
= op
.get('member_vnf_index')
4112 vnf_config_primitive
= op
.get('primitive')
4113 primitive_params
= op
.get('primitive_params')
4114 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
4115 format(vnf_config_primitive
))
4116 # Execute the primitive, either with new (first-time) or registered (reintent) args
4117 ee_descriptor_id
= config_primitive
.get("execution-environment-ref")
4118 primitive_name
= config_primitive
.get("execution-environment-primitive",
4119 vnf_config_primitive
)
4120 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
4121 member_vnf_index
=vnf_index
,
4123 vdu_count_index
=None,
4124 ee_descriptor_id
=ee_descriptor_id
)
4125 result
, result_detail
= await self
._ns
_execute
_primitive
(
4126 ee_id
, primitive_name
, primitive_params
, vca_type
)
4127 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
4128 vnf_config_primitive
, result
, result_detail
))
4129 # Update operationState = COMPLETED | FAILED
4130 self
._update
_suboperation
_status
(
4131 db_nslcmop
, op_index
, result
, result_detail
)
4133 if result
== "FAILED":
4134 raise LcmException(result_detail
)
4135 db_nsr_update
["config-status"] = old_config_status
4136 scale_process
= None
4140 # Should this block be skipped if 'RO_nsr_id' == None ?
4141 # if (RO_nsr_id and RO_scaling_info):
4143 scale_process
= "RO"
4144 # Scale RO retry check: Check if this sub-operation has been executed before
4145 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4146 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
4147 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4148 # Skip sub-operation
4149 result
= 'COMPLETED'
4150 result_detail
= 'Done'
4151 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
4152 result
, result_detail
))
4154 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4155 # New sub-operation: Get index of this sub-operation
4156 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4157 self
.logger
.debug(logging_text
+ "New sub-operation RO")
4159 # retry: Get registered params for this existing sub-operation
4160 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4161 RO_nsr_id
= op
.get('RO_nsr_id')
4162 RO_scaling_info
= op
.get('RO_scaling_info')
4163 self
.logger
.debug(logging_text
+ "Sub-operation RO retry for primitive {}".format(
4164 vnf_config_primitive
))
4166 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
4167 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
4168 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
4170 RO_nslcmop_id
= RO_desc
["instance_action_id"]
4171 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
4173 RO_task_done
= False
4174 step
= detailed_status
= "Waiting for VIM to scale. RO_task_id={}.".format(RO_nslcmop_id
)
4175 detailed_status_old
= None
4176 self
.logger
.debug(logging_text
+ step
)
4178 deployment_timeout
= 1 * 3600 # One hour
4179 while deployment_timeout
> 0:
4180 if not RO_task_done
:
4181 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
4182 extra_item_id
=RO_nslcmop_id
)
4185 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4187 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4188 if ns_status
== "ERROR":
4189 raise ROclient
.ROClientException(ns_status_info
)
4190 elif ns_status
== "BUILD":
4191 detailed_status
= step
+ "; {}".format(ns_status_info
)
4192 elif ns_status
== "ACTIVE":
4194 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
4195 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
4196 self
.logger
.debug(logging_text
+ step
)
4198 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
4200 desc
= await self
.RO
.show("ns", RO_nsr_id
)
4201 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
4203 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4205 if ns_status
== "ERROR":
4206 raise ROclient
.ROClientException(ns_status_info
)
4207 elif ns_status
== "BUILD":
4208 detailed_status
= step
+ "; {}".format(ns_status_info
)
4209 elif ns_status
== "ACTIVE":
4210 step
= detailed_status
= \
4211 "Waiting for management IP address reported by the VIM. Updating VNFRs"
4213 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
4214 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
4216 except LcmExceptionNoMgmtIP
:
4219 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
4220 if detailed_status
!= detailed_status_old
:
4221 self
._update
_suboperation
_status
(
4222 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
4223 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
4224 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
4226 await asyncio
.sleep(5, loop
=self
.loop
)
4227 deployment_timeout
-= 5
4228 if deployment_timeout
<= 0:
4229 self
._update
_suboperation
_status
(
4230 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
4231 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
4233 # update VDU_SCALING_INFO with the obtained ip_addresses
4234 if vdu_scaling_info
["scaling_direction"] == "OUT":
4235 for vdur
in reversed(db_vnfr
["vdur"]):
4236 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
4237 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
4238 vdu_scaling_info
["vdu"].append({
4239 "name": vdur
["name"],
4240 "vdu_id": vdur
["vdu-id-ref"],
4243 for interface
in vdur
["interfaces"]:
4244 vdu_scaling_info
["vdu"][-1]["interface"].append({
4245 "name": interface
["name"],
4246 "ip_address": interface
["ip-address"],
4247 "mac_address": interface
.get("mac-address"),
4249 del vdu_scaling_info
["vdu-create"]
4251 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
4254 scale_process
= None
4256 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4259 # execute primitive service POST-SCALING
4260 step
= "Executing post-scale vnf-config-primitive"
4261 if scaling_descriptor
.get("scaling-config-action"):
4262 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
4263 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
4264 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
4265 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
4266 step
= db_nslcmop_update
["detailed-status"] = \
4267 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
4269 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
4270 if db_vnfr
.get("additionalParamsForVnf"):
4271 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
4273 # look for primitive
4274 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
4275 if config_primitive
["name"] == vnf_config_primitive
:
4279 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
4280 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
4281 "config-primitive".format(scaling_group
, vnf_config_primitive
))
4282 scale_process
= "VCA"
4283 db_nsr_update
["config-status"] = "configuring post-scaling"
4284 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
4286 # Post-scale retry check: Check if this sub-operation has been executed before
4287 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4288 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
4289 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4290 # Skip sub-operation
4291 result
= 'COMPLETED'
4292 result_detail
= 'Done'
4293 self
.logger
.debug(logging_text
+
4294 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
4295 format(vnf_config_primitive
, result
, result_detail
))
4297 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4298 # New sub-operation: Get index of this sub-operation
4299 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4300 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
4301 format(vnf_config_primitive
))
4303 # retry: Get registered params for this existing sub-operation
4304 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4305 vnf_index
= op
.get('member_vnf_index')
4306 vnf_config_primitive
= op
.get('primitive')
4307 primitive_params
= op
.get('primitive_params')
4308 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
4309 format(vnf_config_primitive
))
4310 # Execute the primitive, either with new (first-time) or registered (reintent) args
4311 ee_descriptor_id
= config_primitive
.get("execution-environment-ref")
4312 primitive_name
= config_primitive
.get("execution-environment-primitive",
4313 vnf_config_primitive
)
4314 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
4315 member_vnf_index
=vnf_index
,
4317 vdu_count_index
=None,
4318 ee_descriptor_id
=ee_descriptor_id
)
4319 result
, result_detail
= await self
._ns
_execute
_primitive
(
4320 ee_id
, primitive_name
, primitive_params
, vca_type
)
4321 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
4322 vnf_config_primitive
, result
, result_detail
))
4323 # Update operationState = COMPLETED | FAILED
4324 self
._update
_suboperation
_status
(
4325 db_nslcmop
, op_index
, result
, result_detail
)
4327 if result
== "FAILED":
4328 raise LcmException(result_detail
)
4329 db_nsr_update
["config-status"] = old_config_status
4330 scale_process
= None
4333 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
4334 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
4335 else old_operational_status
4336 db_nsr_update
["config-status"] = old_config_status
4338 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
4339 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4341 except asyncio
.CancelledError
:
4342 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
4343 exc
= "Operation was cancelled"
4344 except Exception as e
:
4345 exc
= traceback
.format_exc()
4346 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
4348 self
._write
_ns
_status
(
4351 current_operation
="IDLE",
4352 current_operation_id
=None
4355 db_nslcmop_update
["detailed-status"] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
4356 nslcmop_operation_state
= "FAILED"
4358 db_nsr_update
["operational-status"] = old_operational_status
4359 db_nsr_update
["config-status"] = old_config_status
4360 db_nsr_update
["detailed-status"] = ""
4362 if "VCA" in scale_process
:
4363 db_nsr_update
["config-status"] = "failed"
4364 if "RO" in scale_process
:
4365 db_nsr_update
["operational-status"] = "failed"
4366 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
4369 error_description_nslcmop
= None
4370 nslcmop_operation_state
= "COMPLETED"
4371 db_nslcmop_update
["detailed-status"] = "Done"
4373 self
._write
_op
_status
(
4376 error_message
=error_description_nslcmop
,
4377 operation_state
=nslcmop_operation_state
,
4378 other_update
=db_nslcmop_update
,
4381 self
._write
_ns
_status
(
4384 current_operation
="IDLE",
4385 current_operation_id
=None,
4386 other_update
=db_nsr_update
4389 if nslcmop_operation_state
:
4391 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
4392 "operationState": nslcmop_operation_state
},
4395 # await asyncio.sleep(cooldown_time, loop=self.loop)
4396 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
4397 except Exception as e
:
4398 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
4399 self
.logger
.debug(logging_text
+ "Exit")
4400 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
4402 async def add_prometheus_metrics(self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
):
4403 if not self
.prometheus
:
4405 # look if exist a file called 'prometheus*.j2' and
4406 artifact_content
= self
.fs
.dir_ls(artifact_path
)
4407 job_file
= next((f
for f
in artifact_content
if f
.startswith("prometheus") and f
.endswith(".j2")), None)
4410 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
4414 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
4415 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
4417 vnfr_id
= vnfr_id
.replace("-", "")
4419 "JOB_NAME": vnfr_id
,
4420 "TARGET_IP": target_ip
,
4421 "EXPORTER_POD_IP": host_name
,
4422 "EXPORTER_POD_PORT": host_port
,
4424 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
4425 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
4426 for job
in job_list
:
4427 if not isinstance(job
.get("job_name"), str) or vnfr_id
not in job
["job_name"]:
4428 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
4429 job
["nsr_id"] = nsr_id
4430 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
4431 if await self
.prometheus
.update(job_dict
):
4432 return list(job_dict
.keys())