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
, Template
, meta
, TemplateError
, TemplateNotFound
, TemplateSyntaxError
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
))
281 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
283 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
284 :param vnfd: input vnfd
285 :param new_id: overrides vnf id if provided
286 :param additionalParams: Instantiation params for VNFs provided
287 :param nsrId: Id of the NSR
288 :return: copy of vnfd
291 vnfd_RO
= deepcopy(vnfd
)
292 # remove unused by RO configuration, monitoring, scaling and internal keys
293 vnfd_RO
.pop("_id", None)
294 vnfd_RO
.pop("_admin", None)
295 vnfd_RO
.pop("vnf-configuration", None)
296 vnfd_RO
.pop("monitoring-param", None)
297 vnfd_RO
.pop("scaling-group-descriptor", None)
298 vnfd_RO
.pop("kdu", None)
299 vnfd_RO
.pop("k8s-cluster", None)
301 vnfd_RO
["id"] = new_id
303 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
304 for vdu
in get_iterable(vnfd_RO
, "vdu"):
305 cloud_init_file
= None
306 if vdu
.get("cloud-init-file"):
307 base_folder
= vnfd
["_admin"]["storage"]
308 cloud_init_file
= "{}/{}/cloud_init/{}".format(base_folder
["folder"], base_folder
["pkg-dir"],
309 vdu
["cloud-init-file"])
310 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
311 cloud_init_content
= ci_file
.read()
312 vdu
.pop("cloud-init-file", None)
313 elif vdu
.get("cloud-init"):
314 cloud_init_content
= vdu
["cloud-init"]
319 ast
= env
.parse(cloud_init_content
)
320 mandatory_vars
= meta
.find_undeclared_variables(ast
)
322 for var
in mandatory_vars
:
323 if not additionalParams
or var
not in additionalParams
.keys():
324 raise LcmException("Variable '{}' defined at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
325 "file, must be provided in the instantiation parameters inside the "
326 "'additionalParamsForVnf' block".format(var
, vnfd
["id"], vdu
["id"]))
327 template
= Template(cloud_init_content
)
328 cloud_init_content
= template
.render(additionalParams
or {})
329 vdu
["cloud-init"] = cloud_init_content
332 except FsException
as e
:
333 raise LcmException("Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".
334 format(vnfd
["id"], vdu
["id"], cloud_init_file
, e
))
335 except (TemplateError
, TemplateNotFound
, TemplateSyntaxError
) as e
:
336 raise LcmException("Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".
337 format(vnfd
["id"], vdu
["id"], e
))
339 def _ns_params_2_RO(self
, ns_params
, nsd
, vnfd_dict
, db_vnfrs
, n2vc_key_list
):
341 Creates a RO ns descriptor from OSM ns_instantiate params
342 :param ns_params: OSM instantiate params
343 :param vnfd_dict: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
344 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index. {member-vnf-index: {vnfr_object}, ...}
345 :return: The RO ns descriptor
349 # TODO feature 1417: Check that no instantiation is set over PDU
350 # check if PDU forces a concrete vim-network-id and add it
351 # check if PDU contains a SDN-assist info (dpid, switch, port) and pass it to RO
353 def vim_account_2_RO(vim_account
):
354 if vim_account
in vim_2_RO
:
355 return vim_2_RO
[vim_account
]
357 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
358 if db_vim
["_admin"]["operationalState"] != "ENABLED":
359 raise LcmException("VIM={} is not available. operationalState={}".format(
360 vim_account
, db_vim
["_admin"]["operationalState"]))
361 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
362 vim_2_RO
[vim_account
] = RO_vim_id
365 def wim_account_2_RO(wim_account
):
366 if isinstance(wim_account
, str):
367 if wim_account
in wim_2_RO
:
368 return wim_2_RO
[wim_account
]
370 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
371 if db_wim
["_admin"]["operationalState"] != "ENABLED":
372 raise LcmException("WIM={} is not available. operationalState={}".format(
373 wim_account
, db_wim
["_admin"]["operationalState"]))
374 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
375 wim_2_RO
[wim_account
] = RO_wim_id
380 def ip_profile_2_RO(ip_profile
):
381 RO_ip_profile
= deepcopy((ip_profile
))
382 if "dns-server" in RO_ip_profile
:
383 if isinstance(RO_ip_profile
["dns-server"], list):
384 RO_ip_profile
["dns-address"] = []
385 for ds
in RO_ip_profile
.pop("dns-server"):
386 RO_ip_profile
["dns-address"].append(ds
['address'])
388 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
389 if RO_ip_profile
.get("ip-version") == "ipv4":
390 RO_ip_profile
["ip-version"] = "IPv4"
391 if RO_ip_profile
.get("ip-version") == "ipv6":
392 RO_ip_profile
["ip-version"] = "IPv6"
393 if "dhcp-params" in RO_ip_profile
:
394 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
400 # "name": ns_params["nsName"],
401 # "description": ns_params.get("nsDescription"),
402 "datacenter": vim_account_2_RO(ns_params
["vimAccountId"]),
403 "wim_account": wim_account_2_RO(ns_params
.get("wimAccountId")),
404 # "scenario": ns_params["nsdId"],
406 # set vim_account of each vnf if different from general vim_account.
407 # Get this information from <vnfr> database content, key vim-account-id
408 # Vim account can be set by placement_engine and it may be different from
409 # the instantiate parameters (vnfs.member-vnf-index.datacenter).
410 for vnf_index
, vnfr
in db_vnfrs
.items():
411 if vnfr
.get("vim-account-id") and vnfr
["vim-account-id"] != ns_params
["vimAccountId"]:
412 populate_dict(RO_ns_params
, ("vnfs", vnf_index
, "datacenter"), vim_account_2_RO(vnfr
["vim-account-id"]))
414 n2vc_key_list
= n2vc_key_list
or []
415 for vnfd_ref
, vnfd
in vnfd_dict
.items():
416 vdu_needed_access
= []
418 if vnfd
.get("vnf-configuration"):
419 ssh_required
= deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required"))
420 if ssh_required
and vnfd
.get("mgmt-interface"):
421 if vnfd
["mgmt-interface"].get("vdu-id"):
422 vdu_needed_access
.append(vnfd
["mgmt-interface"]["vdu-id"])
423 elif vnfd
["mgmt-interface"].get("cp"):
424 mgmt_cp
= vnfd
["mgmt-interface"]["cp"]
426 for vdu
in vnfd
.get("vdu", ()):
427 if vdu
.get("vdu-configuration"):
428 ssh_required
= deep_get(vdu
, ("vdu-configuration", "config-access", "ssh-access", "required"))
430 vdu_needed_access
.append(vdu
["id"])
432 for vdu_interface
in vdu
.get("interface"):
433 if vdu_interface
.get("external-connection-point-ref") and \
434 vdu_interface
["external-connection-point-ref"] == mgmt_cp
:
435 vdu_needed_access
.append(vdu
["id"])
439 if vdu_needed_access
:
440 for vnf_member
in nsd
.get("constituent-vnfd"):
441 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
443 for vdu
in vdu_needed_access
:
444 populate_dict(RO_ns_params
,
445 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
, "mgmt_keys"),
448 if ns_params
.get("vduImage"):
449 RO_ns_params
["vduImage"] = ns_params
["vduImage"]
451 if ns_params
.get("ssh_keys"):
452 RO_ns_params
["cloud-config"] = {"key-pairs": ns_params
["ssh_keys"]}
453 for vnf_params
in get_iterable(ns_params
, "vnf"):
454 for constituent_vnfd
in nsd
["constituent-vnfd"]:
455 if constituent_vnfd
["member-vnf-index"] == vnf_params
["member-vnf-index"]:
456 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
459 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index={} is not present at nsd:"
460 "constituent-vnfd".format(vnf_params
["member-vnf-index"]))
462 for vdu_params
in get_iterable(vnf_params
, "vdu"):
463 # TODO feature 1417: check that this VDU exist and it is not a PDU
464 if vdu_params
.get("volume"):
465 for volume_params
in vdu_params
["volume"]:
466 if volume_params
.get("vim-volume-id"):
467 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
468 vdu_params
["id"], "devices", volume_params
["name"], "vim_id"),
469 volume_params
["vim-volume-id"])
470 if vdu_params
.get("interface"):
471 for interface_params
in vdu_params
["interface"]:
472 if interface_params
.get("ip-address"):
473 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
474 vdu_params
["id"], "interfaces", interface_params
["name"],
476 interface_params
["ip-address"])
477 if interface_params
.get("mac-address"):
478 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
479 vdu_params
["id"], "interfaces", interface_params
["name"],
481 interface_params
["mac-address"])
482 if interface_params
.get("floating-ip-required"):
483 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
484 vdu_params
["id"], "interfaces", interface_params
["name"],
486 interface_params
["floating-ip-required"])
488 for internal_vld_params
in get_iterable(vnf_params
, "internal-vld"):
489 if internal_vld_params
.get("vim-network-name"):
490 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
491 internal_vld_params
["name"], "vim-network-name"),
492 internal_vld_params
["vim-network-name"])
493 if internal_vld_params
.get("vim-network-id"):
494 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
495 internal_vld_params
["name"], "vim-network-id"),
496 internal_vld_params
["vim-network-id"])
497 if internal_vld_params
.get("ip-profile"):
498 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
499 internal_vld_params
["name"], "ip-profile"),
500 ip_profile_2_RO(internal_vld_params
["ip-profile"]))
501 if internal_vld_params
.get("provider-network"):
503 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
504 internal_vld_params
["name"], "provider-network"),
505 internal_vld_params
["provider-network"].copy())
507 for icp_params
in get_iterable(internal_vld_params
, "internal-connection-point"):
510 for vdu_descriptor
in vnf_descriptor
["vdu"]:
511 for vdu_interface
in vdu_descriptor
["interface"]:
512 if vdu_interface
.get("internal-connection-point-ref") == icp_params
["id-ref"]:
513 if icp_params
.get("ip-address"):
514 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
515 vdu_descriptor
["id"], "interfaces",
516 vdu_interface
["name"], "ip_address"),
517 icp_params
["ip-address"])
519 if icp_params
.get("mac-address"):
520 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
521 vdu_descriptor
["id"], "interfaces",
522 vdu_interface
["name"], "mac_address"),
523 icp_params
["mac-address"])
529 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index[{}]:"
530 "internal-vld:id-ref={} is not present at vnfd:internal-"
531 "connection-point".format(vnf_params
["member-vnf-index"],
532 icp_params
["id-ref"]))
534 for vld_params
in get_iterable(ns_params
, "vld"):
535 if "ip-profile" in vld_params
:
536 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "ip-profile"),
537 ip_profile_2_RO(vld_params
["ip-profile"]))
539 if vld_params
.get("provider-network"):
541 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "provider-network"),
542 vld_params
["provider-network"].copy())
544 if "wimAccountId" in vld_params
and vld_params
["wimAccountId"] is not None:
545 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "wim_account"),
546 wim_account_2_RO(vld_params
["wimAccountId"])),
547 if vld_params
.get("vim-network-name"):
549 if isinstance(vld_params
["vim-network-name"], dict):
550 for vim_account
, vim_net
in vld_params
["vim-network-name"].items():
551 RO_vld_sites
.append({
552 "netmap-use": vim_net
,
553 "datacenter": vim_account_2_RO(vim_account
)
555 else: # isinstance str
556 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-name"]})
558 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
560 if vld_params
.get("vim-network-id"):
562 if isinstance(vld_params
["vim-network-id"], dict):
563 for vim_account
, vim_net
in vld_params
["vim-network-id"].items():
564 RO_vld_sites
.append({
565 "netmap-use": vim_net
,
566 "datacenter": vim_account_2_RO(vim_account
)
568 else: # isinstance str
569 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-id"]})
571 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
572 if vld_params
.get("ns-net"):
573 if isinstance(vld_params
["ns-net"], dict):
574 for vld_id
, instance_scenario_id
in vld_params
["ns-net"].items():
575 RO_vld_ns_net
= {"instance_scenario_id": instance_scenario_id
, "osm_id": vld_id
}
576 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "use-network"), RO_vld_ns_net
)
577 if "vnfd-connection-point-ref" in vld_params
:
578 for cp_params
in vld_params
["vnfd-connection-point-ref"]:
580 for constituent_vnfd
in nsd
["constituent-vnfd"]:
581 if constituent_vnfd
["member-vnf-index"] == cp_params
["member-vnf-index-ref"]:
582 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
586 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={} "
587 "is not present at nsd:constituent-vnfd".format(cp_params
["member-vnf-index-ref"]))
589 for vdu_descriptor
in vnf_descriptor
["vdu"]:
590 for interface_descriptor
in vdu_descriptor
["interface"]:
591 if interface_descriptor
.get("external-connection-point-ref") == \
592 cp_params
["vnfd-connection-point-ref"]:
599 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={}:"
600 "vnfd-connection-point-ref={} is not present at vnfd={}".format(
601 cp_params
["member-vnf-index-ref"],
602 cp_params
["vnfd-connection-point-ref"],
603 vnf_descriptor
["id"]))
604 if cp_params
.get("ip-address"):
605 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
606 vdu_descriptor
["id"], "interfaces",
607 interface_descriptor
["name"], "ip_address"),
608 cp_params
["ip-address"])
609 if cp_params
.get("mac-address"):
610 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
611 vdu_descriptor
["id"], "interfaces",
612 interface_descriptor
["name"], "mac_address"),
613 cp_params
["mac-address"])
616 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None):
617 # make a copy to do not change
618 vdu_create
= copy(vdu_create
)
619 vdu_delete
= copy(vdu_delete
)
621 vdurs
= db_vnfr
.get("vdur")
624 vdu_index
= len(vdurs
)
627 vdur
= vdurs
[vdu_index
]
628 if vdur
.get("pdu-type"):
630 vdu_id_ref
= vdur
["vdu-id-ref"]
631 if vdu_create
and vdu_create
.get(vdu_id_ref
):
632 vdur_copy
= deepcopy(vdur
)
633 vdur_copy
["status"] = "BUILD"
634 vdur_copy
["status-detailed"] = None
635 vdur_copy
["ip_address"]: None
636 for iface
in vdur_copy
["interfaces"]:
637 iface
["ip-address"] = None
638 iface
["mac-address"] = None
639 iface
.pop("mgmt_vnf", None) # only first vdu can be managment of vnf # TODO ALF
640 for index
in range(0, vdu_create
[vdu_id_ref
]):
641 vdur_copy
["_id"] = str(uuid4())
642 vdur_copy
["count-index"] += 1
643 vdurs
.insert(vdu_index
+1+index
, vdur_copy
)
644 self
.logger
.debug("scale out, adding vdu={}".format(vdur_copy
))
645 vdur_copy
= deepcopy(vdur_copy
)
647 del vdu_create
[vdu_id_ref
]
648 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
650 vdu_delete
[vdu_id_ref
] -= 1
651 if not vdu_delete
[vdu_id_ref
]:
652 del vdu_delete
[vdu_id_ref
]
653 # check all operations are done
654 if vdu_create
or vdu_delete
:
655 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
658 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
661 vnfr_update
= {"vdur": vdurs
}
662 db_vnfr
["vdur"] = vdurs
663 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
665 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
667 Updates database nsr with the RO info for the created vld
668 :param ns_update_nsr: dictionary to be filled with the updated info
669 :param db_nsr: content of db_nsr. This is also modified
670 :param nsr_desc_RO: nsr descriptor from RO
671 :return: Nothing, LcmException is raised on errors
674 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
675 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
676 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
678 vld
["vim-id"] = net_RO
.get("vim_net_id")
679 vld
["name"] = net_RO
.get("vim_name")
680 vld
["status"] = net_RO
.get("status")
681 vld
["status-detailed"] = net_RO
.get("error_msg")
682 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
685 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
687 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
689 for db_vnfr
in db_vnfrs
.values():
690 vnfr_update
= {"status": "ERROR"}
691 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
692 if "status" not in vdur
:
693 vdur
["status"] = "ERROR"
694 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
696 vdur
["status-detailed"] = str(error_text
)
697 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
698 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
699 except DbException
as e
:
700 self
.logger
.error("Cannot update vnf. {}".format(e
))
702 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
704 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
705 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
706 :param nsr_desc_RO: nsr descriptor from RO
707 :return: Nothing, LcmException is raised on errors
709 for vnf_index
, db_vnfr
in db_vnfrs
.items():
710 for vnf_RO
in nsr_desc_RO
["vnfs"]:
711 if vnf_RO
["member_vnf_index"] != vnf_index
:
714 if vnf_RO
.get("ip_address"):
715 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
716 elif not db_vnfr
.get("ip-address"):
717 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
718 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
720 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
721 vdur_RO_count_index
= 0
722 if vdur
.get("pdu-type"):
724 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
725 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
727 if vdur
["count-index"] != vdur_RO_count_index
:
728 vdur_RO_count_index
+= 1
730 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
731 if vdur_RO
.get("ip_address"):
732 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
734 vdur
["ip-address"] = None
735 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
736 vdur
["name"] = vdur_RO
.get("vim_name")
737 vdur
["status"] = vdur_RO
.get("status")
738 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
739 for ifacer
in get_iterable(vdur
, "interfaces"):
740 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
741 if ifacer
["name"] == interface_RO
.get("internal_name"):
742 ifacer
["ip-address"] = interface_RO
.get("ip_address")
743 ifacer
["mac-address"] = interface_RO
.get("mac_address")
746 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
748 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
749 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
752 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
753 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
755 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
756 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
757 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
759 vld
["vim-id"] = net_RO
.get("vim_net_id")
760 vld
["name"] = net_RO
.get("vim_name")
761 vld
["status"] = net_RO
.get("status")
762 vld
["status-detailed"] = net_RO
.get("error_msg")
763 vnfr_update
["vld.{}".format(vld_index
)] = vld
766 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
767 vnf_index
, vld
["id"]))
769 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
773 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
775 def _get_ns_config_info(self
, nsr_id
):
777 Generates a mapping between vnf,vdu elements and the N2VC id
778 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
779 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
780 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
781 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
783 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
784 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
786 ns_config_info
= {"osm-config-mapping": mapping
}
787 for vca
in vca_deployed_list
:
788 if not vca
["member-vnf-index"]:
790 if not vca
["vdu_id"]:
791 mapping
[vca
["member-vnf-index"]] = vca
["application"]
793 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
795 return ns_config_info
798 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
, ee_descriptor_id
):
800 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
801 primitives as verify-ssh-credentials, or config when needed
802 :param desc_primitive_list: information of the descriptor
803 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
804 this element contains a ssh public key
805 :param ee_descriptor_id: execution environment descriptor id. It is the value of
806 XXX_configuration.execution-environment-list.INDEX.id; it can be None
807 :return: The modified list. Can ba an empty list, but always a list
810 primitive_list
= desc_primitive_list
or []
812 # filter primitives by ee_id
813 primitive_list
= [p
for p
in primitive_list
if p
.get("execution-environment-ref") == ee_descriptor_id
]
817 primitive_list
.sort(key
=lambda val
: int(val
['seq']))
819 # look for primitive config, and get the position. None if not present
820 config_position
= None
821 for index
, primitive
in enumerate(primitive_list
):
822 if primitive
["name"] == "config":
823 config_position
= index
826 # for NS, add always a config primitive if not present (bug 874)
827 if not vca_deployed
["member-vnf-index"] and config_position
is None:
828 primitive_list
.insert(0, {"name": "config", "parameter": []})
830 # TODO revise if needed: for VNF/VDU add verify-ssh-credentials after config
831 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
832 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
833 return primitive_list
835 async def _instantiate_ng_ro(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
836 n2vc_key_list
, stage
, start_deploy
, timeout_ns_deploy
):
837 nslcmop_id
= db_nslcmop
["_id"]
839 "name": db_nsr
["name"],
842 "image": deepcopy(db_nsr
["image"]),
843 "flavor": deepcopy(db_nsr
["flavor"]),
844 "action_id": nslcmop_id
,
846 for image
in target
["image"]:
847 image
["vim_info"] = []
848 for flavor
in target
["flavor"]:
849 flavor
["vim_info"] = []
851 ns_params
= db_nslcmop
.get("operationParams")
853 if ns_params
.get("ssh_keys"):
854 ssh_keys
+= ns_params
.get("ssh_keys")
856 ssh_keys
+= n2vc_key_list
859 for vld_index
, vld
in enumerate(nsd
.get("vld")):
860 target_vld
= {"id": vld
["id"],
862 "mgmt-network": vld
.get("mgmt-network", False),
863 "type": vld
.get("type"),
864 "vim_info": [{"vim-network-name": vld
.get("vim-network-name"),
865 "vim_account_id": ns_params
["vimAccountId"]}],
867 for cp
in vld
["vnfd-connection-point-ref"]:
868 cp2target
["member_vnf:{}.{}".format(cp
["member-vnf-index-ref"], cp
["vnfd-connection-point-ref"])] = \
869 "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
870 target
["ns"]["vld"].append(target_vld
)
871 for vnfr
in db_vnfrs
.values():
872 vnfd
= db_vnfds_ref
[vnfr
["vnfd-ref"]]
873 target_vnf
= deepcopy(vnfr
)
874 for vld
in target_vnf
.get("vld", ()):
875 # check if connected to a ns.vld
876 vnf_cp
= next((cp
for cp
in vnfd
.get("connection-point", ()) if
877 cp
.get("internal-vld-ref") == vld
["id"]), None)
879 ns_cp
= "member_vnf:{}.{}".format(vnfr
["member-vnf-index-ref"], vnf_cp
["id"])
880 if cp2target
.get(ns_cp
):
881 vld
["target"] = cp2target
[ns_cp
]
882 vld
["vim_info"] = [{"vim-network-name": vld
.get("vim-network-name"),
883 "vim_account_id": vnfr
["vim-account-id"]}]
885 for vdur
in target_vnf
.get("vdur", ()):
886 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
887 vdud_index
, vdud
= next(k
for k
in enumerate(vnfd
["vdu"]) if k
[1]["id"] == vdur
["vdu-id-ref"])
888 # vdur["additionalParams"] = vnfr.get("additionalParamsForVnf") # TODO additional params for VDU
891 if deep_get(vdud
, ("vdu-configuration", "config-access", "ssh-access", "required")):
892 vdur
["ssh-keys"] = ssh_keys
893 vdur
["ssh-access-required"] = True
894 elif deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required")) and \
895 any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"]):
896 vdur
["ssh-keys"] = ssh_keys
897 vdur
["ssh-access-required"] = True
900 if vdud
.get("cloud-init-file"):
901 vdur
["cloud-init"] = "{}:file:{}".format(vnfd
["_id"], vdud
.get("cloud-init-file"))
902 elif vdud
.get("cloud-init"):
903 vdur
["cloud-init"] = "{}:vdu:{}".format(vnfd
["_id"], vdud_index
)
906 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
907 if not next((vi
for vi
in ns_flavor
["vim_info"] if
908 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
909 ns_flavor
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
911 ns_image
= target
["image"][int(vdur
["ns-image-id"])]
912 if not next((vi
for vi
in ns_image
["vim_info"] if
913 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
914 ns_image
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
916 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
917 target
["vnf"].append(target_vnf
)
919 desc
= await self
.RO
.deploy(nsr_id
, target
)
920 action_id
= desc
["action_id"]
921 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
)
925 "_admin.deployed.RO.operational-status": "running",
926 "detailed-status": " ".join(stage
)
928 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
929 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
930 self
._write
_op
_status
(nslcmop_id
, stage
)
931 self
.logger
.debug(logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
))
934 async def _wait_ng_ro(self
, nsr_id
, action_id
, nslcmop_id
, start_time
, timeout
, stage
):
935 detailed_status_old
= None
937 while time() <= start_time
+ timeout
:
938 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
939 if desc_status
["status"] == "FAILED":
940 raise NgRoException(desc_status
["details"])
941 elif desc_status
["status"] == "BUILD":
942 stage
[2] = "VIM: ({})".format(desc_status
["details"])
943 elif desc_status
["status"] == "DONE":
944 stage
[2] = "Deployed at VIM"
947 assert False, "ROclient.check_ns_status returns unknown {}".format(desc_status
["status"])
948 if stage
[2] != detailed_status_old
:
949 detailed_status_old
= stage
[2]
950 db_nsr_update
["detailed-status"] = " ".join(stage
)
951 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
952 self
._write
_op
_status
(nslcmop_id
, stage
)
953 await asyncio
.sleep(5, loop
=self
.loop
)
954 else: # timeout_ns_deploy
955 raise NgRoException("Timeout waiting ns to deploy")
957 async def _terminate_ng_ro(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
961 start_deploy
= time()
969 desc
= await self
.RO
.deploy(nsr_id
, target
)
970 action_id
= desc
["action_id"]
971 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
972 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
973 self
.logger
.debug(logging_text
+ "ns terminate action at RO. action_id={}".format(action_id
))
976 delete_timeout
= 20 * 60 # 20 minutes
977 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
)
979 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
980 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
982 await self
.RO
.delete(nsr_id
)
983 except Exception as e
:
984 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
985 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
986 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
987 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
988 self
.logger
.debug(logging_text
+ "RO_action_id={} already deleted".format(action_id
))
989 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
990 failed_detail
.append("delete conflict: {}".format(e
))
991 self
.logger
.debug(logging_text
+ "RO_action_id={} delete conflict: {}".format(action_id
, e
))
993 failed_detail
.append("delete error: {}".format(e
))
994 self
.logger
.error(logging_text
+ "RO_action_id={} delete error: {}".format(action_id
, e
))
997 stage
[2] = "Error deleting from VIM"
999 stage
[2] = "Deleted from VIM"
1000 db_nsr_update
["detailed-status"] = " ".join(stage
)
1001 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1002 self
._write
_op
_status
(nslcmop_id
, stage
)
1005 raise LcmException("; ".join(failed_detail
))
1008 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
1009 n2vc_key_list
, stage
):
1012 :param logging_text: preffix text to use at logging
1013 :param nsr_id: nsr identity
1014 :param nsd: database content of ns descriptor
1015 :param db_nsr: database content of ns record
1016 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1018 :param db_vnfds_ref: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1019 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1020 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1021 :return: None or exception
1025 RO_descriptor_number
= 0 # number of descriptors created at RO
1026 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
1027 nslcmop_id
= db_nslcmop
["_id"]
1028 start_deploy
= time()
1029 ns_params
= db_nslcmop
.get("operationParams")
1030 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1031 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1033 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1035 # Check for and optionally request placement optimization. Database will be updated if placement activated
1036 stage
[2] = "Waiting for Placement."
1037 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1038 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1039 for vnfr
in db_vnfrs
.values():
1040 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1043 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1046 return await self
._instantiate
_ng
_ro
(logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
,
1047 db_vnfds_ref
, n2vc_key_list
, stage
, start_deploy
,
1050 # get vnfds, instantiate at RO
1051 for c_vnf
in nsd
.get("constituent-vnfd", ()):
1052 member_vnf_index
= c_vnf
["member-vnf-index"]
1053 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
1054 vnfd_ref
= vnfd
["id"]
1056 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
1057 db_nsr_update
["detailed-status"] = " ".join(stage
)
1058 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1059 self
._write
_op
_status
(nslcmop_id
, stage
)
1061 # self.logger.debug(logging_text + stage[2])
1062 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
1063 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
1064 RO_descriptor_number
+= 1
1066 # look position at deployed.RO.vnfd if not present it will be appended at the end
1067 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
1068 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
1071 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
1072 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
1075 RO_update
= {"member-vnf-index": member_vnf_index
}
1076 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
1078 RO_update
["id"] = vnfd_list
[0]["uuid"]
1079 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
1080 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
1082 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
1083 get("additionalParamsForVnf"), nsr_id
)
1084 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
1085 RO_update
["id"] = desc
["uuid"]
1086 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
1087 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
1088 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
1089 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
1094 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1095 db_nsr_update
["detailed-status"] = " ".join(stage
)
1096 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1097 self
._write
_op
_status
(nslcmop_id
, stage
)
1099 # self.logger.debug(logging_text + stage[2])
1100 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
1101 RO_descriptor_number
+= 1
1102 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
1104 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
1105 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
1106 nsd_ref
, RO_nsd_uuid
))
1108 nsd_RO
= deepcopy(nsd
)
1109 nsd_RO
["id"] = RO_osm_nsd_id
1110 nsd_RO
.pop("_id", None)
1111 nsd_RO
.pop("_admin", None)
1112 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
1113 member_vnf_index
= c_vnf
["member-vnf-index"]
1114 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1115 for c_vld
in nsd_RO
.get("vld", ()):
1116 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
1117 member_vnf_index
= cp
["member-vnf-index-ref"]
1118 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1120 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
1121 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1122 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
1123 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
1124 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1127 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1128 db_nsr_update
["detailed-status"] = " ".join(stage
)
1129 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1130 self
._write
_op
_status
(nslcmop_id
, stage
)
1132 # if present use it unless in error status
1133 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
1136 stage
[2] = "Looking for existing ns at RO"
1137 db_nsr_update
["detailed-status"] = " ".join(stage
)
1138 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1139 self
._write
_op
_status
(nslcmop_id
, stage
)
1140 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
1141 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1143 except ROclient
.ROClientException
as e
:
1144 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
1146 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1148 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1149 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1150 if ns_status
== "ERROR":
1151 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
1152 self
.logger
.debug(logging_text
+ stage
[2])
1153 await self
.RO
.delete("ns", RO_nsr_id
)
1154 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1156 stage
[2] = "Checking dependencies"
1157 db_nsr_update
["detailed-status"] = " ".join(stage
)
1158 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1159 self
._write
_op
_status
(nslcmop_id
, stage
)
1160 # self.logger.debug(logging_text + stage[2])
1162 # check if VIM is creating and wait look if previous tasks in process
1163 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
1165 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
1166 self
.logger
.debug(logging_text
+ stage
[2])
1167 await asyncio
.wait(task_dependency
, timeout
=3600)
1168 if ns_params
.get("vnf"):
1169 for vnf
in ns_params
["vnf"]:
1170 if "vimAccountId" in vnf
:
1171 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
1172 vnf
["vimAccountId"])
1174 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
1175 self
.logger
.debug(logging_text
+ stage
[2])
1176 await asyncio
.wait(task_dependency
, timeout
=3600)
1178 stage
[2] = "Checking instantiation parameters."
1179 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, db_vnfrs
, n2vc_key_list
)
1180 stage
[2] = "Deploying ns at VIM."
1181 db_nsr_update
["detailed-status"] = " ".join(stage
)
1182 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1183 self
._write
_op
_status
(nslcmop_id
, stage
)
1185 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
1186 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
1187 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1188 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
1189 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
1191 # wait until NS is ready
1192 stage
[2] = "Waiting VIM to deploy ns."
1193 db_nsr_update
["detailed-status"] = " ".join(stage
)
1194 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1195 self
._write
_op
_status
(nslcmop_id
, stage
)
1196 detailed_status_old
= None
1197 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
1200 while time() <= start_deploy
+ timeout_ns_deploy
:
1201 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1204 if desc
!= old_desc
:
1205 # desc has changed => update db
1206 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
1209 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1210 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1211 if ns_status
== "ERROR":
1212 raise ROclient
.ROClientException(ns_status_info
)
1213 elif ns_status
== "BUILD":
1214 stage
[2] = "VIM: ({})".format(ns_status_info
)
1215 elif ns_status
== "ACTIVE":
1216 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
1218 self
.ns_update_vnfr(db_vnfrs
, desc
)
1220 except LcmExceptionNoMgmtIP
:
1223 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
1224 if stage
[2] != detailed_status_old
:
1225 detailed_status_old
= stage
[2]
1226 db_nsr_update
["detailed-status"] = " ".join(stage
)
1227 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1228 self
._write
_op
_status
(nslcmop_id
, stage
)
1229 await asyncio
.sleep(5, loop
=self
.loop
)
1230 else: # timeout_ns_deploy
1231 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
1234 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
1236 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
1237 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1238 stage
[2] = "Deployed at VIM"
1239 db_nsr_update
["detailed-status"] = " ".join(stage
)
1240 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1241 self
._write
_op
_status
(nslcmop_id
, stage
)
1242 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
1243 # self.logger.debug(logging_text + "Deployed at VIM")
1244 except (ROclient
.ROClientException
, LcmException
, DbException
, NgRoException
) as e
:
1245 stage
[2] = "ERROR deploying at VIM"
1246 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1249 async def wait_kdu_up(self
, logging_text
, nsr_id
, vnfr_id
, kdu_name
):
1251 Wait for kdu to be up, get ip address
1252 :param logging_text: prefix use for logging
1259 # self.logger.debug(logging_text + "Starting wait_kdu_up")
1262 while nb_tries
< 360:
1263 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1264 kdur
= next((x
for x
in get_iterable(db_vnfr
, "kdur") if x
.get("name") == kdu_name
), None)
1266 raise LcmException("Not found vnfr_id={}, kdu_name={}".format(vnfr_id
, kdu_name
))
1267 if kdur
.get("status"):
1268 if kdur
["status"] in ("READY", "ENABLED"):
1269 return kdur
.get("ip-address")
1271 raise LcmException("target KDU={} is in error state".format(kdu_name
))
1273 await asyncio
.sleep(10, loop
=self
.loop
)
1275 raise LcmException("Timeout waiting KDU={} instantiated".format(kdu_name
))
1277 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
1279 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1280 :param logging_text: prefix use for logging
1285 :param pub_key: public ssh key to inject, None to skip
1286 :param user: user to apply the public ssh key
1290 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
1294 target_vdu_id
= None
1300 if ro_retries
>= 360: # 1 hour
1301 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
1303 await asyncio
.sleep(10, loop
=self
.loop
)
1306 if not target_vdu_id
:
1307 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1309 if not vdu_id
: # for the VNF case
1310 if db_vnfr
.get("status") == "ERROR":
1311 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1312 ip_address
= db_vnfr
.get("ip-address")
1315 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1317 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1318 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1320 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1321 vdur
= db_vnfr
["vdur"][0]
1323 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1326 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1327 ip_address
= vdur
.get("ip-address")
1330 target_vdu_id
= vdur
["vdu-id-ref"]
1331 elif vdur
.get("status") == "ERROR":
1332 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1334 if not target_vdu_id
:
1337 # inject public key into machine
1338 if pub_key
and user
:
1339 # wait until NS is deployed at RO
1341 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1342 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1346 # self.logger.debug(logging_text + "Inserting RO key")
1347 if vdur
.get("pdu-type"):
1348 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1351 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1353 target
= {"action": "inject_ssh_key", "key": pub_key
, "user": user
,
1354 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdu_id
}]}],
1356 await self
.RO
.deploy(nsr_id
, target
)
1358 result_dict
= await self
.RO
.create_action(
1360 item_id_name
=ro_nsr_id
,
1361 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1363 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1364 if not result_dict
or not isinstance(result_dict
, dict):
1365 raise LcmException("Unknown response from RO when injecting key")
1366 for result
in result_dict
.values():
1367 if result
.get("vim_result") == 200:
1370 raise ROclient
.ROClientException("error injecting key: {}".format(
1371 result
.get("description")))
1373 except NgRoException
as e
:
1374 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1375 except ROclient
.ROClientException
as e
:
1377 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1381 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1387 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1389 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1391 my_vca
= vca_deployed_list
[vca_index
]
1392 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1393 # vdu or kdu: no dependencies
1397 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1398 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1399 configuration_status_list
= db_nsr
["configurationStatus"]
1400 for index
, vca_deployed
in enumerate(configuration_status_list
):
1401 if index
== vca_index
:
1404 if not my_vca
.get("member-vnf-index") or \
1405 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1406 internal_status
= configuration_status_list
[index
].get("status")
1407 if internal_status
== 'READY':
1409 elif internal_status
== 'BROKEN':
1410 raise LcmException("Configuration aborted because dependent charm/s has failed")
1414 # no dependencies, return
1416 await asyncio
.sleep(10)
1419 raise LcmException("Configuration aborted because dependent charm/s timeout")
1421 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1422 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
, vca_type
, vca_name
,
1423 ee_config_descriptor
):
1424 nsr_id
= db_nsr
["_id"]
1425 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1426 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1427 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1428 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1430 'collection': 'nsrs',
1431 'filter': {'_id': nsr_id
},
1432 'path': db_update_entry
1438 element_under_configuration
= nsr_id
1442 vnfr_id
= db_vnfr
["_id"]
1443 osm_config
["osm"]["vnf_id"] = vnfr_id
1445 namespace
= "{nsi}.{ns}".format(
1446 nsi
=nsi_id
if nsi_id
else "",
1450 element_type
= 'VNF'
1451 element_under_configuration
= vnfr_id
1452 namespace
+= ".{}".format(vnfr_id
)
1454 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1455 element_type
= 'VDU'
1456 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1457 osm_config
["osm"]["vdu_id"] = vdu_id
1459 namespace
+= ".{}".format(kdu_name
)
1460 element_type
= 'KDU'
1461 element_under_configuration
= kdu_name
1462 osm_config
["osm"]["kdu_name"] = kdu_name
1465 artifact_path
= "{}/{}/{}/{}".format(
1466 base_folder
["folder"],
1467 base_folder
["pkg-dir"],
1468 "charms" if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") else "helm-charts",
1471 # get initial_config_primitive_list that applies to this element
1472 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1474 # add config if not present for NS charm
1475 ee_descriptor_id
= ee_config_descriptor
.get("id")
1476 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1477 vca_deployed
, ee_descriptor_id
)
1479 # n2vc_redesign STEP 3.1
1480 # find old ee_id if exists
1481 ee_id
= vca_deployed
.get("ee_id")
1483 # create or register execution environment in VCA
1484 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm"):
1486 self
._write
_configuration
_status
(
1488 vca_index
=vca_index
,
1490 element_under_configuration
=element_under_configuration
,
1491 element_type
=element_type
1494 step
= "create execution environment"
1495 self
.logger
.debug(logging_text
+ step
)
1496 ee_id
, credentials
= await self
.vca_map
[vca_type
].create_execution_environment(
1497 namespace
=namespace
,
1501 artifact_path
=artifact_path
,
1504 elif vca_type
== "native_charm":
1505 step
= "Waiting to VM being up and getting IP address"
1506 self
.logger
.debug(logging_text
+ step
)
1507 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1508 user
=None, pub_key
=None)
1509 credentials
= {"hostname": rw_mgmt_ip
}
1511 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1512 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1513 # merged. Meanwhile let's get username from initial-config-primitive
1514 if not username
and initial_config_primitive_list
:
1515 for config_primitive
in initial_config_primitive_list
:
1516 for param
in config_primitive
.get("parameter", ()):
1517 if param
["name"] == "ssh-username":
1518 username
= param
["value"]
1521 raise LcmException("Cannot determine the username neither with 'initial-config-primitive' nor with "
1522 "'config-access.ssh-access.default-user'")
1523 credentials
["username"] = username
1524 # n2vc_redesign STEP 3.2
1526 self
._write
_configuration
_status
(
1528 vca_index
=vca_index
,
1529 status
='REGISTERING',
1530 element_under_configuration
=element_under_configuration
,
1531 element_type
=element_type
1534 step
= "register execution environment {}".format(credentials
)
1535 self
.logger
.debug(logging_text
+ step
)
1536 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1537 credentials
=credentials
, namespace
=namespace
, db_dict
=db_dict
)
1539 # for compatibility with MON/POL modules, the need model and application name at database
1540 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1541 ee_id_parts
= ee_id
.split('.')
1542 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1543 if len(ee_id_parts
) >= 2:
1544 model_name
= ee_id_parts
[0]
1545 application_name
= ee_id_parts
[1]
1546 db_nsr_update
[db_update_entry
+ "model"] = model_name
1547 db_nsr_update
[db_update_entry
+ "application"] = application_name
1549 # n2vc_redesign STEP 3.3
1550 step
= "Install configuration Software"
1552 self
._write
_configuration
_status
(
1554 vca_index
=vca_index
,
1555 status
='INSTALLING SW',
1556 element_under_configuration
=element_under_configuration
,
1557 element_type
=element_type
,
1558 other_update
=db_nsr_update
1561 # TODO check if already done
1562 self
.logger
.debug(logging_text
+ step
)
1564 if vca_type
== "native_charm":
1565 config_primitive
= next((p
for p
in initial_config_primitive_list
if p
["name"] == "config"), None)
1566 if config_primitive
:
1567 config
= self
._map
_primitive
_params
(
1573 if vca_type
== "lxc_proxy_charm":
1574 if element_type
== "NS":
1575 num_units
= db_nsr
.get("config-units") or 1
1576 elif element_type
== "VNF":
1577 num_units
= db_vnfr
.get("config-units") or 1
1578 elif element_type
== "VDU":
1579 for v
in db_vnfr
["vdur"]:
1580 if vdu_id
== v
["vdu-id-ref"]:
1581 num_units
= v
.get("config-units") or 1
1584 await self
.vca_map
[vca_type
].install_configuration_sw(
1586 artifact_path
=artifact_path
,
1589 num_units
=num_units
,
1593 # write in db flag of configuration_sw already installed
1594 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1596 # add relations for this VCA (wait for other peers related with this VCA)
1597 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
,
1598 vca_index
=vca_index
, vca_type
=vca_type
)
1600 # if SSH access is required, then get execution environment SSH public
1601 # if native charm we have waited already to VM be UP
1602 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm"):
1605 # self.logger.debug("get ssh key block")
1606 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1607 # self.logger.debug("ssh key needed")
1608 # Needed to inject a ssh key
1609 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1610 step
= "Install configuration Software, getting public ssh key"
1611 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1613 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1615 # self.logger.debug("no need to get ssh key")
1616 step
= "Waiting to VM being up and getting IP address"
1617 self
.logger
.debug(logging_text
+ step
)
1619 # n2vc_redesign STEP 5.1
1620 # wait for RO (ip-address) Insert pub_key into VM
1623 rw_mgmt_ip
= await self
.wait_kdu_up(logging_text
, nsr_id
, vnfr_id
, kdu_name
)
1625 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
,
1626 vdu_index
, user
=user
, pub_key
=pub_key
)
1628 rw_mgmt_ip
= None # This is for a NS configuration
1630 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1632 # store rw_mgmt_ip in deploy params for later replacement
1633 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1635 # n2vc_redesign STEP 6 Execute initial config primitive
1636 step
= 'execute initial config primitive'
1638 # wait for dependent primitives execution (NS -> VNF -> VDU)
1639 if initial_config_primitive_list
:
1640 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1642 # stage, in function of element type: vdu, kdu, vnf or ns
1643 my_vca
= vca_deployed_list
[vca_index
]
1644 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1646 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1647 elif my_vca
.get("member-vnf-index"):
1649 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1652 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1654 self
._write
_configuration
_status
(
1656 vca_index
=vca_index
,
1657 status
='EXECUTING PRIMITIVE'
1660 self
._write
_op
_status
(
1665 check_if_terminated_needed
= True
1666 for initial_config_primitive
in initial_config_primitive_list
:
1667 # adding information on the vca_deployed if it is a NS execution environment
1668 if not vca_deployed
["member-vnf-index"]:
1669 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1670 # TODO check if already done
1671 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1673 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1674 self
.logger
.debug(logging_text
+ step
)
1675 await self
.vca_map
[vca_type
].exec_primitive(
1677 primitive_name
=initial_config_primitive
["name"],
1678 params_dict
=primitive_params_
,
1681 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1682 if check_if_terminated_needed
:
1683 if config_descriptor
.get('terminate-config-primitive'):
1684 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1685 check_if_terminated_needed
= False
1687 # TODO register in database that primitive is done
1689 # STEP 7 Configure metrics
1690 if vca_type
== "helm":
1691 prometheus_jobs
= await self
.add_prometheus_metrics(
1693 artifact_path
=artifact_path
,
1694 ee_config_descriptor
=ee_config_descriptor
,
1697 target_ip
=rw_mgmt_ip
,
1700 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "prometheus_jobs": prometheus_jobs
})
1702 step
= "instantiated at VCA"
1703 self
.logger
.debug(logging_text
+ step
)
1705 self
._write
_configuration
_status
(
1707 vca_index
=vca_index
,
1711 except Exception as e
: # TODO not use Exception but N2VC exception
1712 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1713 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1714 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1715 self
._write
_configuration
_status
(
1717 vca_index
=vca_index
,
1720 raise LcmException("{} {}".format(step
, e
)) from e
1722 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1723 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1725 Update db_nsr fields.
1728 :param current_operation:
1729 :param current_operation_id:
1730 :param error_description:
1731 :param error_detail:
1732 :param other_update: Other required changes at database if provided, will be cleared
1736 db_dict
= other_update
or {}
1737 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1738 db_dict
["_admin.current-operation"] = current_operation_id
1739 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1740 db_dict
["currentOperation"] = current_operation
1741 db_dict
["currentOperationID"] = current_operation_id
1742 db_dict
["errorDescription"] = error_description
1743 db_dict
["errorDetail"] = error_detail
1746 db_dict
["nsState"] = ns_state
1747 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1748 except DbException
as e
:
1749 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1751 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1752 operation_state
: str = None, other_update
: dict = None):
1754 db_dict
= other_update
or {}
1755 db_dict
['queuePosition'] = queuePosition
1756 if isinstance(stage
, list):
1757 db_dict
['stage'] = stage
[0]
1758 db_dict
['detailed-status'] = " ".join(stage
)
1759 elif stage
is not None:
1760 db_dict
['stage'] = str(stage
)
1762 if error_message
is not None:
1763 db_dict
['errorMessage'] = error_message
1764 if operation_state
is not None:
1765 db_dict
['operationState'] = operation_state
1766 db_dict
["statusEnteredTime"] = time()
1767 self
.update_db_2("nslcmops", op_id
, db_dict
)
1768 except DbException
as e
:
1769 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1771 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1773 nsr_id
= db_nsr
["_id"]
1774 # configurationStatus
1775 config_status
= db_nsr
.get('configurationStatus')
1777 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1778 enumerate(config_status
) if v
}
1780 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1782 except DbException
as e
:
1783 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1785 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1786 element_under_configuration
: str = None, element_type
: str = None,
1787 other_update
: dict = None):
1789 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1790 # .format(vca_index, status))
1793 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1794 db_dict
= other_update
or {}
1796 db_dict
[db_path
+ 'status'] = status
1797 if element_under_configuration
:
1798 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1800 db_dict
[db_path
+ 'elementType'] = element_type
1801 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1802 except DbException
as e
:
1803 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1804 .format(status
, nsr_id
, vca_index
, e
))
1806 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1808 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1809 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1810 Database is used because the result can be obtained from a different LCM worker in case of HA.
1811 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1812 :param db_nslcmop: database content of nslcmop
1813 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1814 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
1815 computed 'vim-account-id'
1818 nslcmop_id
= db_nslcmop
['_id']
1819 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1820 if placement_engine
== "PLA":
1821 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1822 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1823 db_poll_interval
= 5
1824 wait
= db_poll_interval
* 10
1826 while not pla_result
and wait
>= 0:
1827 await asyncio
.sleep(db_poll_interval
)
1828 wait
-= db_poll_interval
1829 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1830 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1833 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1835 for pla_vnf
in pla_result
['vnf']:
1836 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1837 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1840 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1842 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1845 def update_nsrs_with_pla_result(self
, params
):
1847 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1848 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1849 except Exception as e
:
1850 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1852 async def instantiate(self
, nsr_id
, nslcmop_id
):
1855 :param nsr_id: ns instance to deploy
1856 :param nslcmop_id: operation to run
1860 # Try to lock HA task here
1861 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1862 if not task_is_locked_by_me
:
1863 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1866 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1867 self
.logger
.debug(logging_text
+ "Enter")
1869 # get all needed from database
1871 # database nsrs record
1874 # database nslcmops record
1877 # update operation on nsrs
1879 # update operation on nslcmops
1880 db_nslcmop_update
= {}
1882 nslcmop_operation_state
= None
1883 db_vnfrs
= {} # vnf's info indexed by member-index
1885 tasks_dict_info
= {} # from task to info text
1888 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1889 # ^ stage, step, VIM progress
1891 # wait for any previous tasks in process
1892 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1894 stage
[1] = "Sync filesystem from database."
1895 self
.fs
.sync() # TODO, make use of partial sync, only for the needed packages
1897 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1898 stage
[1] = "Reading from database."
1899 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1900 db_nsr_update
["detailed-status"] = "creating"
1901 db_nsr_update
["operational-status"] = "init"
1902 self
._write
_ns
_status
(
1904 ns_state
="BUILDING",
1905 current_operation
="INSTANTIATING",
1906 current_operation_id
=nslcmop_id
,
1907 other_update
=db_nsr_update
1909 self
._write
_op
_status
(
1915 # read from db: operation
1916 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
1917 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1918 ns_params
= db_nslcmop
.get("operationParams")
1919 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1920 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1922 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1925 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
1926 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1927 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
1928 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
1930 # nsr_name = db_nsr["name"] # TODO short-name??
1932 # read from db: vnf's of this ns
1933 stage
[1] = "Getting vnfrs from db."
1934 self
.logger
.debug(logging_text
+ stage
[1])
1935 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1937 # read from db: vnfd's for every vnf
1938 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1939 db_vnfds
= {} # every vnfd data indexed by vnf id
1940 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1942 # for each vnf in ns, read vnfd
1943 for vnfr
in db_vnfrs_list
:
1944 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1945 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1946 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1948 # if we haven't this vnfd, read it from db
1949 if vnfd_id
not in db_vnfds
:
1951 stage
[1] = "Getting vnfd={} id='{}' from db.".format(vnfd_id
, vnfd_ref
)
1952 self
.logger
.debug(logging_text
+ stage
[1])
1953 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1956 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1957 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1958 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1960 # Get or generates the _admin.deployed.VCA list
1961 vca_deployed_list
= None
1962 if db_nsr
["_admin"].get("deployed"):
1963 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1964 if vca_deployed_list
is None:
1965 vca_deployed_list
= []
1966 configuration_status_list
= []
1967 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1968 db_nsr_update
["configurationStatus"] = configuration_status_list
1969 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1970 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1971 elif isinstance(vca_deployed_list
, dict):
1972 # maintain backward compatibility. Change a dict to list at database
1973 vca_deployed_list
= list(vca_deployed_list
.values())
1974 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1975 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1977 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1978 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1979 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1981 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1982 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1983 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1984 self
.db
.set_list("vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"})
1986 # n2vc_redesign STEP 2 Deploy Network Scenario
1987 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1988 self
._write
_op
_status
(
1993 stage
[1] = "Deploying KDUs."
1994 # self.logger.debug(logging_text + "Before deploy_kdus")
1995 # Call to deploy_kdus in case exists the "vdu:kdu" param
1996 await self
.deploy_kdus(
1997 logging_text
=logging_text
,
1999 nslcmop_id
=nslcmop_id
,
2002 task_instantiation_info
=tasks_dict_info
,
2005 stage
[1] = "Getting VCA public key."
2006 # n2vc_redesign STEP 1 Get VCA public ssh-key
2007 # feature 1429. Add n2vc public key to needed VMs
2008 n2vc_key
= self
.n2vc
.get_public_key()
2009 n2vc_key_list
= [n2vc_key
]
2010 if self
.vca_config
.get("public_key"):
2011 n2vc_key_list
.append(self
.vca_config
["public_key"])
2013 stage
[1] = "Deploying NS at VIM."
2014 task_ro
= asyncio
.ensure_future(
2015 self
.instantiate_RO(
2016 logging_text
=logging_text
,
2020 db_nslcmop
=db_nslcmop
,
2022 db_vnfds_ref
=db_vnfds_ref
,
2023 n2vc_key_list
=n2vc_key_list
,
2027 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
2028 tasks_dict_info
[task_ro
] = "Deploying at VIM"
2030 # n2vc_redesign STEP 3 to 6 Deploy N2VC
2031 stage
[1] = "Deploying Execution Environments."
2032 self
.logger
.debug(logging_text
+ stage
[1])
2034 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
2035 # get_iterable() returns a value from a dict or empty tuple if key does not exist
2036 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
2037 vnfd_id
= c_vnf
["vnfd-id-ref"]
2038 vnfd
= db_vnfds_ref
[vnfd_id
]
2039 member_vnf_index
= str(c_vnf
["member-vnf-index"])
2040 db_vnfr
= db_vnfrs
[member_vnf_index
]
2041 base_folder
= vnfd
["_admin"]["storage"]
2047 # Get additional parameters
2049 if db_vnfr
.get("additionalParamsForVnf"):
2050 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
2052 descriptor_config
= vnfd
.get("vnf-configuration")
2053 if descriptor_config
:
2055 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
2058 nslcmop_id
=nslcmop_id
,
2064 member_vnf_index
=member_vnf_index
,
2065 vdu_index
=vdu_index
,
2067 deploy_params
=deploy_params
,
2068 descriptor_config
=descriptor_config
,
2069 base_folder
=base_folder
,
2070 task_instantiation_info
=tasks_dict_info
,
2074 # Deploy charms for each VDU that supports one.
2075 for vdud
in get_iterable(vnfd
, 'vdu'):
2077 descriptor_config
= vdud
.get('vdu-configuration')
2078 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
2079 if vdur
.get("additionalParams"):
2080 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
2082 deploy_params_vdu
= deploy_params
2083 if descriptor_config
:
2084 # look for vdu index in the db_vnfr["vdu"] section
2085 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
2086 # if vdur["vdu-id-ref"] == vdu_id:
2089 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
2090 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
2091 # vdu_name = vdur.get("name")
2094 for vdu_index
in range(int(vdud
.get("count", 1))):
2095 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2097 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2098 member_vnf_index
, vdu_id
, vdu_index
),
2101 nslcmop_id
=nslcmop_id
,
2107 member_vnf_index
=member_vnf_index
,
2108 vdu_index
=vdu_index
,
2110 deploy_params
=deploy_params_vdu
,
2111 descriptor_config
=descriptor_config
,
2112 base_folder
=base_folder
,
2113 task_instantiation_info
=tasks_dict_info
,
2116 for kdud
in get_iterable(vnfd
, 'kdu'):
2117 kdu_name
= kdud
["name"]
2118 descriptor_config
= kdud
.get('kdu-configuration')
2119 if descriptor_config
:
2123 # look for vdu index in the db_vnfr["vdu"] section
2124 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
2125 # if vdur["vdu-id-ref"] == vdu_id:
2128 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
2129 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
2130 # vdu_name = vdur.get("name")
2134 logging_text
=logging_text
,
2137 nslcmop_id
=nslcmop_id
,
2143 member_vnf_index
=member_vnf_index
,
2144 vdu_index
=vdu_index
,
2146 deploy_params
=deploy_params
,
2147 descriptor_config
=descriptor_config
,
2148 base_folder
=base_folder
,
2149 task_instantiation_info
=tasks_dict_info
,
2153 # Check if this NS has a charm configuration
2154 descriptor_config
= nsd
.get("ns-configuration")
2155 if descriptor_config
and descriptor_config
.get("juju"):
2158 member_vnf_index
= None
2164 # Get additional parameters
2166 if db_nsr
.get("additionalParamsForNs"):
2167 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
2168 base_folder
= nsd
["_admin"]["storage"]
2170 logging_text
=logging_text
,
2173 nslcmop_id
=nslcmop_id
,
2179 member_vnf_index
=member_vnf_index
,
2180 vdu_index
=vdu_index
,
2182 deploy_params
=deploy_params
,
2183 descriptor_config
=descriptor_config
,
2184 base_folder
=base_folder
,
2185 task_instantiation_info
=tasks_dict_info
,
2189 # rest of staff will be done at finally
2191 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2192 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
2194 except asyncio
.CancelledError
:
2195 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2196 exc
= "Operation was cancelled"
2197 except Exception as e
:
2198 exc
= traceback
.format_exc()
2199 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2202 error_list
.append(str(exc
))
2204 # wait for pending tasks
2206 stage
[1] = "Waiting for instantiate pending tasks."
2207 self
.logger
.debug(logging_text
+ stage
[1])
2208 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
2209 stage
, nslcmop_id
, nsr_id
=nsr_id
)
2210 stage
[1] = stage
[2] = ""
2211 except asyncio
.CancelledError
:
2212 error_list
.append("Cancelled")
2213 # TODO cancel all tasks
2214 except Exception as exc
:
2215 error_list
.append(str(exc
))
2217 # update operation-status
2218 db_nsr_update
["operational-status"] = "running"
2219 # let's begin with VCA 'configured' status (later we can change it)
2220 db_nsr_update
["config-status"] = "configured"
2221 for task
, task_name
in tasks_dict_info
.items():
2222 if not task
.done() or task
.cancelled() or task
.exception():
2223 if task_name
.startswith(self
.task_name_deploy_vca
):
2224 # A N2VC task is pending
2225 db_nsr_update
["config-status"] = "failed"
2227 # RO or KDU task is pending
2228 db_nsr_update
["operational-status"] = "failed"
2230 # update status at database
2232 error_detail
= ". ".join(error_list
)
2233 self
.logger
.error(logging_text
+ error_detail
)
2234 error_description_nslcmop
= '{} Detail: {}'.format(stage
[0], error_detail
)
2235 error_description_nsr
= 'Operation: INSTANTIATING.{}, {}'.format(nslcmop_id
, stage
[0])
2237 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2238 db_nslcmop_update
["detailed-status"] = error_detail
2239 nslcmop_operation_state
= "FAILED"
2243 error_description_nsr
= error_description_nslcmop
= None
2245 db_nsr_update
["detailed-status"] = "Done"
2246 db_nslcmop_update
["detailed-status"] = "Done"
2247 nslcmop_operation_state
= "COMPLETED"
2250 self
._write
_ns
_status
(
2253 current_operation
="IDLE",
2254 current_operation_id
=None,
2255 error_description
=error_description_nsr
,
2256 error_detail
=error_detail
,
2257 other_update
=db_nsr_update
2259 self
._write
_op
_status
(
2262 error_message
=error_description_nslcmop
,
2263 operation_state
=nslcmop_operation_state
,
2264 other_update
=db_nslcmop_update
,
2267 if nslcmop_operation_state
:
2269 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2270 "operationState": nslcmop_operation_state
},
2272 except Exception as e
:
2273 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2275 self
.logger
.debug(logging_text
+ "Exit")
2276 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2278 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int,
2279 timeout
: int = 3600, vca_type
: str = None) -> bool:
2282 # 1. find all relations for this VCA
2283 # 2. wait for other peers related
2287 vca_type
= vca_type
or "lxc_proxy_charm"
2289 # STEP 1: find all relations for this VCA
2292 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2293 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2296 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
2298 # read all ns-configuration relations
2299 ns_relations
= list()
2300 db_ns_relations
= deep_get(nsd
, ('ns-configuration', 'relation'))
2302 for r
in db_ns_relations
:
2303 # check if this VCA is in the relation
2304 if my_vca
.get('member-vnf-index') in\
2305 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2306 ns_relations
.append(r
)
2308 # read all vnf-configuration relations
2309 vnf_relations
= list()
2310 db_vnfd_list
= db_nsr
.get('vnfd-id')
2312 for vnfd
in db_vnfd_list
:
2313 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2314 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
2315 if db_vnf_relations
:
2316 for r
in db_vnf_relations
:
2317 # check if this VCA is in the relation
2318 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2319 vnf_relations
.append(r
)
2321 # if no relations, terminate
2322 if not ns_relations
and not vnf_relations
:
2323 self
.logger
.debug(logging_text
+ ' No relations')
2326 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
2333 if now
- start
>= timeout
:
2334 self
.logger
.error(logging_text
+ ' : timeout adding relations')
2337 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2338 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2340 # for each defined NS relation, find the VCA's related
2341 for r
in ns_relations
:
2342 from_vca_ee_id
= None
2344 from_vca_endpoint
= None
2345 to_vca_endpoint
= None
2346 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2347 for vca
in vca_list
:
2348 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
2349 and vca
.get('config_sw_installed'):
2350 from_vca_ee_id
= vca
.get('ee_id')
2351 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2352 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2353 and vca
.get('config_sw_installed'):
2354 to_vca_ee_id
= vca
.get('ee_id')
2355 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2356 if from_vca_ee_id
and to_vca_ee_id
:
2358 await self
.vca_map
[vca_type
].add_relation(
2359 ee_id_1
=from_vca_ee_id
,
2360 ee_id_2
=to_vca_ee_id
,
2361 endpoint_1
=from_vca_endpoint
,
2362 endpoint_2
=to_vca_endpoint
)
2363 # remove entry from relations list
2364 ns_relations
.remove(r
)
2366 # check failed peers
2368 vca_status_list
= db_nsr
.get('configurationStatus')
2370 for i
in range(len(vca_list
)):
2372 vca_status
= vca_status_list
[i
]
2373 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2374 if vca_status
.get('status') == 'BROKEN':
2375 # peer broken: remove relation from list
2376 ns_relations
.remove(r
)
2377 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2378 if vca_status
.get('status') == 'BROKEN':
2379 # peer broken: remove relation from list
2380 ns_relations
.remove(r
)
2385 # for each defined VNF relation, find the VCA's related
2386 for r
in vnf_relations
:
2387 from_vca_ee_id
= None
2389 from_vca_endpoint
= None
2390 to_vca_endpoint
= None
2391 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2392 for vca
in vca_list
:
2393 key_to_check
= "vdu_id"
2394 if vca
.get("vdu_id") is None:
2395 key_to_check
= "vnfd_id"
2396 if vca
.get(key_to_check
) == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2397 from_vca_ee_id
= vca
.get('ee_id')
2398 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2399 if vca
.get(key_to_check
) == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2400 to_vca_ee_id
= vca
.get('ee_id')
2401 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2402 if from_vca_ee_id
and to_vca_ee_id
:
2404 await self
.vca_map
[vca_type
].add_relation(
2405 ee_id_1
=from_vca_ee_id
,
2406 ee_id_2
=to_vca_ee_id
,
2407 endpoint_1
=from_vca_endpoint
,
2408 endpoint_2
=to_vca_endpoint
)
2409 # remove entry from relations list
2410 vnf_relations
.remove(r
)
2412 # check failed peers
2414 vca_status_list
= db_nsr
.get('configurationStatus')
2416 for i
in range(len(vca_list
)):
2418 vca_status
= vca_status_list
[i
]
2419 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2420 if vca_status
.get('status') == 'BROKEN':
2421 # peer broken: remove relation from list
2422 ns_relations
.remove(r
)
2423 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2424 if vca_status
.get('status') == 'BROKEN':
2425 # peer broken: remove relation from list
2426 ns_relations
.remove(r
)
2432 await asyncio
.sleep(5.0)
2434 if not ns_relations
and not vnf_relations
:
2435 self
.logger
.debug('Relations added')
2440 except Exception as e
:
2441 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2444 async def _install_kdu(self
, nsr_id
: str, nsr_db_path
: str, vnfr_data
: dict, kdu_index
: int, kdud
: dict,
2445 vnfd
: dict, k8s_instance_info
: dict, k8params
: dict = None, timeout
: int = 600):
2448 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
2450 db_dict_install
= {"collection": "nsrs",
2451 "filter": {"_id": nsr_id
},
2452 "path": nsr_db_path
}
2454 kdu_instance
= await self
.k8scluster_map
[k8sclustertype
].install(
2455 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2456 kdu_model
=k8s_instance_info
["kdu-model"],
2459 db_dict
=db_dict_install
,
2461 kdu_name
=k8s_instance_info
["kdu-name"],
2462 namespace
=k8s_instance_info
["namespace"])
2463 self
.update_db_2("nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
})
2465 # Obtain services to obtain management service ip
2466 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
2467 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2468 kdu_instance
=kdu_instance
,
2469 namespace
=k8s_instance_info
["namespace"])
2471 # Obtain management service info (if exists)
2472 vnfr_update_dict
= {}
2474 vnfr_update_dict
["kdur.{}.services".format(kdu_index
)] = services
2475 mgmt_services
= [service
for service
in kdud
.get("service", []) if service
.get("mgmt-service")]
2476 for mgmt_service
in mgmt_services
:
2477 for service
in services
:
2478 if service
["name"].startswith(mgmt_service
["name"]):
2479 # Mgmt service found, Obtain service ip
2480 ip
= service
.get("external_ip", service
.get("cluster_ip"))
2481 if isinstance(ip
, list) and len(ip
) == 1:
2484 vnfr_update_dict
["kdur.{}.ip-address".format(kdu_index
)] = ip
2486 # Check if must update also mgmt ip at the vnf
2487 service_external_cp
= mgmt_service
.get("external-connection-point-ref")
2488 if service_external_cp
:
2489 if deep_get(vnfd
, ("mgmt-interface", "cp")) == service_external_cp
:
2490 vnfr_update_dict
["ip-address"] = ip
2494 self
.logger
.warn("Mgmt service name: {} not found".format(mgmt_service
["name"]))
2496 vnfr_update_dict
["kdur.{}.status".format(kdu_index
)] = "READY"
2497 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
2499 kdu_config
= kdud
.get("kdu-configuration")
2500 if kdu_config
and kdu_config
.get("initial-config-primitive") and kdu_config
.get("juju") is None:
2501 initial_config_primitive_list
= kdu_config
.get("initial-config-primitive")
2502 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
2504 for initial_config_primitive
in initial_config_primitive_list
:
2505 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, {})
2507 await asyncio
.wait_for(
2508 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
2509 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2510 kdu_instance
=kdu_instance
,
2511 primitive_name
=initial_config_primitive
["name"],
2512 params
=primitive_params_
, db_dict
={}),
2515 except Exception as e
:
2516 # Prepare update db with error and raise exception
2518 self
.update_db_2("nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)})
2519 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), {"kdur.{}.status".format(kdu_index
): "ERROR"})
2521 # ignore to keep original exception
2523 # reraise original error
2528 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2529 # Launch kdus if present in the descriptor
2531 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2533 async def _get_cluster_id(cluster_id
, cluster_type
):
2534 nonlocal k8scluster_id_2_uuic
2535 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2536 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2538 # check if K8scluster is creating and wait look if previous tasks in process
2539 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("k8scluster", cluster_id
)
2541 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(task_name
, cluster_id
)
2542 self
.logger
.debug(logging_text
+ text
)
2543 await asyncio
.wait(task_dependency
, timeout
=3600)
2545 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2546 if not db_k8scluster
:
2547 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2549 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2551 raise LcmException("K8s cluster '{}' has not been initialized for '{}'".format(cluster_id
,
2553 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2556 logging_text
+= "Deploy kdus: "
2559 db_nsr_update
= {"_admin.deployed.K8s": []}
2560 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2563 updated_cluster_list
= []
2565 for vnfr_data
in db_vnfrs
.values():
2566 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
2567 # Step 0: Prepare and set parameters
2568 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2569 vnfd_id
= vnfr_data
.get('vnfd-id')
2570 kdud
= next(kdud
for kdud
in db_vnfds
[vnfd_id
]["kdu"] if kdud
["name"] == kdur
["kdu-name"])
2571 namespace
= kdur
.get("k8s-namespace")
2572 if kdur
.get("helm-chart"):
2573 kdumodel
= kdur
["helm-chart"]
2574 k8sclustertype
= "helm-chart"
2575 elif kdur
.get("juju-bundle"):
2576 kdumodel
= kdur
["juju-bundle"]
2577 k8sclustertype
= "juju-bundle"
2579 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2580 "juju-bundle. Maybe an old NBI version is running".
2581 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2582 # check if kdumodel is a file and exists
2584 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2585 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2586 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2587 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["pkg-dir"], k8sclustertype
,
2589 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2590 kdumodel
= self
.fs
.path
+ filename
2591 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2593 except Exception: # it is not a file
2596 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2597 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2598 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2601 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2602 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2603 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2604 if del_repo_list
or added_repo_dict
:
2605 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2606 updated
= {'_admin.helm_charts_added.' +
2607 item
: name
for item
, name
in added_repo_dict
.items()}
2608 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2609 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2611 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2612 updated_cluster_list
.append(cluster_uuid
)
2615 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2616 kdur
["kdu-name"], k8s_cluster_id
)
2617 k8s_instance_info
= {"kdu-instance": None,
2618 "k8scluster-uuid": cluster_uuid
,
2619 "k8scluster-type": k8sclustertype
,
2620 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2621 "kdu-name": kdur
["kdu-name"],
2622 "kdu-model": kdumodel
,
2623 "namespace": namespace
}
2624 db_path
= "_admin.deployed.K8s.{}".format(index
)
2625 db_nsr_update
[db_path
] = k8s_instance_info
2626 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2628 task
= asyncio
.ensure_future(
2629 self
._install
_kdu
(nsr_id
, db_path
, vnfr_data
, kdu_index
, kdud
, db_vnfds
[vnfd_id
],
2630 k8s_instance_info
, k8params
=desc_params
, timeout
=600))
2631 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2632 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2636 except (LcmException
, asyncio
.CancelledError
):
2638 except Exception as e
:
2639 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2640 if isinstance(e
, (N2VCException
, DbException
)):
2641 self
.logger
.error(logging_text
+ msg
)
2643 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2644 raise LcmException(msg
)
2647 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2649 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2650 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2651 base_folder
, task_instantiation_info
, stage
):
2652 # launch instantiate_N2VC in a asyncio task and register task object
2653 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2654 # if not found, create one entry and update database
2655 # fill db_nsr._admin.deployed.VCA.<index>
2657 self
.logger
.debug(logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
))
2658 if descriptor_config
.get("juju"): # There is one execution envioronment of type juju
2659 ee_list
= [descriptor_config
]
2660 elif descriptor_config
.get("execution-environment-list"):
2661 ee_list
= descriptor_config
.get("execution-environment-list")
2662 else: # other types as script are not supported
2665 for ee_item
in ee_list
:
2666 self
.logger
.debug(logging_text
+ "_deploy_n2vc ee_item juju={}, helm={}".format(ee_item
.get('juju'),
2667 ee_item
.get("helm-chart")))
2668 ee_descriptor_id
= ee_item
.get("id")
2669 if ee_item
.get("juju"):
2670 vca_name
= ee_item
['juju'].get('charm')
2671 vca_type
= "lxc_proxy_charm" if ee_item
['juju'].get('charm') is not None else "native_charm"
2672 if ee_item
['juju'].get('cloud') == "k8s":
2673 vca_type
= "k8s_proxy_charm"
2674 elif ee_item
['juju'].get('proxy') is False:
2675 vca_type
= "native_charm"
2676 elif ee_item
.get("helm-chart"):
2677 vca_name
= ee_item
['helm-chart']
2680 self
.logger
.debug(logging_text
+ "skipping non juju neither charm configuration")
2684 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2685 if not vca_deployed
:
2687 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2688 vca_deployed
.get("vdu_id") == vdu_id
and \
2689 vca_deployed
.get("kdu_name") == kdu_name
and \
2690 vca_deployed
.get("vdu_count_index", 0) == vdu_index
and \
2691 vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
:
2694 # not found, create one.
2695 target
= "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
2697 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
2699 target
+= "/kdu/{}".format(kdu_name
)
2701 "target_element": target
,
2702 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
2703 "member-vnf-index": member_vnf_index
,
2705 "kdu_name": kdu_name
,
2706 "vdu_count_index": vdu_index
,
2707 "operational-status": "init", # TODO revise
2708 "detailed-status": "", # TODO revise
2709 "step": "initial-deploy", # TODO revise
2711 "vdu_name": vdu_name
,
2713 "ee_descriptor_id": ee_descriptor_id
2717 # create VCA and configurationStatus in db
2719 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2720 "configurationStatus.{}".format(vca_index
): dict()
2722 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2724 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2727 task_n2vc
= asyncio
.ensure_future(
2728 self
.instantiate_N2VC(
2729 logging_text
=logging_text
,
2730 vca_index
=vca_index
,
2736 vdu_index
=vdu_index
,
2737 deploy_params
=deploy_params
,
2738 config_descriptor
=descriptor_config
,
2739 base_folder
=base_folder
,
2740 nslcmop_id
=nslcmop_id
,
2744 ee_config_descriptor
=ee_item
2747 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2748 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2749 member_vnf_index
or "", vdu_id
or "")
2752 def _get_terminate_config_primitive(primitive_list
, vca_deployed
):
2753 """ Get a sorted terminate config primitive list. In case ee_descriptor_id is present at vca_deployed,
2754 it get only those primitives for this execution envirom"""
2756 primitive_list
= primitive_list
or []
2757 # filter primitives by ee_descriptor_id
2758 ee_descriptor_id
= vca_deployed
.get("ee_descriptor_id")
2759 primitive_list
= [p
for p
in primitive_list
if p
.get("execution-environment-ref") == ee_descriptor_id
]
2762 primitive_list
.sort(key
=lambda val
: int(val
['seq']))
2764 return primitive_list
2767 def _create_nslcmop(nsr_id
, operation
, params
):
2769 Creates a ns-lcm-opp content to be stored at database.
2770 :param nsr_id: internal id of the instance
2771 :param operation: instantiate, terminate, scale, action, ...
2772 :param params: user parameters for the operation
2773 :return: dictionary following SOL005 format
2775 # Raise exception if invalid arguments
2776 if not (nsr_id
and operation
and params
):
2778 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2784 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2785 "operationState": "PROCESSING",
2786 "statusEnteredTime": now
,
2787 "nsInstanceId": nsr_id
,
2788 "lcmOperationType": operation
,
2790 "isAutomaticInvocation": False,
2791 "operationParams": params
,
2792 "isCancelPending": False,
2794 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2795 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2800 def _format_additional_params(self
, params
):
2801 params
= params
or {}
2802 for key
, value
in params
.items():
2803 if str(value
).startswith("!!yaml "):
2804 params
[key
] = yaml
.safe_load(value
[7:])
2807 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2808 primitive
= seq
.get('name')
2809 primitive_params
= {}
2811 "member_vnf_index": vnf_index
,
2812 "primitive": primitive
,
2813 "primitive_params": primitive_params
,
2816 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2820 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2821 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2822 if op
.get('operationState') == 'COMPLETED':
2823 # b. Skip sub-operation
2824 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2825 return self
.SUBOPERATION_STATUS_SKIP
2827 # c. retry executing sub-operation
2828 # The sub-operation exists, and operationState != 'COMPLETED'
2829 # Update operationState = 'PROCESSING' to indicate a retry.
2830 operationState
= 'PROCESSING'
2831 detailed_status
= 'In progress'
2832 self
._update
_suboperation
_status
(
2833 db_nslcmop
, op_index
, operationState
, detailed_status
)
2834 # Return the sub-operation index
2835 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2836 # with arguments extracted from the sub-operation
2839 # Find a sub-operation where all keys in a matching dictionary must match
2840 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2841 def _find_suboperation(self
, db_nslcmop
, match
):
2842 if db_nslcmop
and match
:
2843 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2844 for i
, op
in enumerate(op_list
):
2845 if all(op
.get(k
) == match
[k
] for k
in match
):
2847 return self
.SUBOPERATION_STATUS_NOT_FOUND
2849 # Update status for a sub-operation given its index
2850 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2851 # Update DB for HA tasks
2852 q_filter
= {'_id': db_nslcmop
['_id']}
2853 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2854 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2855 self
.db
.set_one("nslcmops",
2857 update_dict
=update_dict
,
2858 fail_on_empty
=False)
2860 # Add sub-operation, return the index of the added sub-operation
2861 # Optionally, set operationState, detailed-status, and operationType
2862 # Status and type are currently set for 'scale' sub-operations:
2863 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2864 # 'detailed-status' : status message
2865 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2866 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2867 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2868 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2869 RO_nsr_id
=None, RO_scaling_info
=None):
2871 return self
.SUBOPERATION_STATUS_NOT_FOUND
2872 # Get the "_admin.operations" list, if it exists
2873 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2874 op_list
= db_nslcmop_admin
.get('operations')
2875 # Create or append to the "_admin.operations" list
2876 new_op
= {'member_vnf_index': vnf_index
,
2878 'vdu_count_index': vdu_count_index
,
2879 'primitive': primitive
,
2880 'primitive_params': mapped_primitive_params
}
2882 new_op
['operationState'] = operationState
2884 new_op
['detailed-status'] = detailed_status
2886 new_op
['lcmOperationType'] = operationType
2888 new_op
['RO_nsr_id'] = RO_nsr_id
2890 new_op
['RO_scaling_info'] = RO_scaling_info
2892 # No existing operations, create key 'operations' with current operation as first list element
2893 db_nslcmop_admin
.update({'operations': [new_op
]})
2894 op_list
= db_nslcmop_admin
.get('operations')
2896 # Existing operations, append operation to list
2897 op_list
.append(new_op
)
2899 db_nslcmop_update
= {'_admin.operations': op_list
}
2900 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2901 op_index
= len(op_list
) - 1
2904 # Helper methods for scale() sub-operations
2906 # pre-scale/post-scale:
2907 # Check for 3 different cases:
2908 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2909 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2910 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2911 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2912 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2913 # Find this sub-operation
2914 if RO_nsr_id
and RO_scaling_info
:
2915 operationType
= 'SCALE-RO'
2917 'member_vnf_index': vnf_index
,
2918 'RO_nsr_id': RO_nsr_id
,
2919 'RO_scaling_info': RO_scaling_info
,
2923 'member_vnf_index': vnf_index
,
2924 'primitive': vnf_config_primitive
,
2925 'primitive_params': primitive_params
,
2926 'lcmOperationType': operationType
2928 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2929 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2930 # a. New sub-operation
2931 # The sub-operation does not exist, add it.
2932 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2933 # The following parameters are set to None for all kind of scaling:
2935 vdu_count_index
= None
2937 if RO_nsr_id
and RO_scaling_info
:
2938 vnf_config_primitive
= None
2939 primitive_params
= None
2942 RO_scaling_info
= None
2943 # Initial status for sub-operation
2944 operationState
= 'PROCESSING'
2945 detailed_status
= 'In progress'
2946 # Add sub-operation for pre/post-scaling (zero or more operations)
2947 self
._add
_suboperation
(db_nslcmop
,
2952 vnf_config_primitive
,
2959 return self
.SUBOPERATION_STATUS_NEW
2961 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2962 # or op_index (operationState != 'COMPLETED')
2963 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2965 # Function to return execution_environment id
2967 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2968 # TODO vdu_index_count
2969 for vca
in vca_deployed_list
:
2970 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2973 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
,
2974 vca_index
, destroy_ee
=True, exec_primitives
=True):
2976 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2977 :param logging_text:
2979 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2980 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2981 :param vca_index: index in the database _admin.deployed.VCA
2982 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2983 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
2984 not executed properly
2985 :return: None or exception
2989 logging_text
+ " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
2990 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
2994 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
2996 # execute terminate_primitives
2998 terminate_primitives
= self
._get
_terminate
_config
_primitive
(
2999 config_descriptor
.get("terminate-config-primitive"), vca_deployed
)
3000 vdu_id
= vca_deployed
.get("vdu_id")
3001 vdu_count_index
= vca_deployed
.get("vdu_count_index")
3002 vdu_name
= vca_deployed
.get("vdu_name")
3003 vnf_index
= vca_deployed
.get("member-vnf-index")
3004 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
3005 for seq
in terminate_primitives
:
3006 # For each sequence in list, get primitive and call _ns_execute_primitive()
3007 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
3008 vnf_index
, seq
.get("name"))
3009 self
.logger
.debug(logging_text
+ step
)
3010 # Create the primitive for each sequence, i.e. "primitive": "touch"
3011 primitive
= seq
.get('name')
3012 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
3015 self
._add
_suboperation
(db_nslcmop
,
3021 mapped_primitive_params
)
3022 # Sub-operations: Call _ns_execute_primitive() instead of action()
3024 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
3025 mapped_primitive_params
,
3027 except LcmException
:
3028 # this happens when VCA is not deployed. In this case it is not needed to terminate
3030 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
3031 if result
not in result_ok
:
3032 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
3033 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
3034 # set that this VCA do not need terminated
3035 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
3036 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
3038 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
3039 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
3042 await self
.vca_map
[vca_type
].delete_execution_environment(vca_deployed
["ee_id"])
3044 async def _delete_all_N2VC(self
, db_nsr
: dict):
3045 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
3046 namespace
= "." + db_nsr
["_id"]
3048 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
3049 except N2VCNotFound
: # already deleted. Skip
3051 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
3053 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
3055 Terminates a deployment from RO
3056 :param logging_text:
3057 :param nsr_deployed: db_nsr._admin.deployed
3060 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3061 this method will update only the index 2, but it will write on database the concatenated content of the list
3066 ro_nsr_id
= ro_delete_action
= None
3067 if nsr_deployed
and nsr_deployed
.get("RO"):
3068 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3069 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3072 stage
[2] = "Deleting ns from VIM."
3073 db_nsr_update
["detailed-status"] = " ".join(stage
)
3074 self
._write
_op
_status
(nslcmop_id
, stage
)
3075 self
.logger
.debug(logging_text
+ stage
[2])
3076 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3077 self
._write
_op
_status
(nslcmop_id
, stage
)
3078 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3079 ro_delete_action
= desc
["action_id"]
3080 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
3081 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3082 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3083 if ro_delete_action
:
3084 # wait until NS is deleted from VIM
3085 stage
[2] = "Waiting ns deleted from VIM."
3086 detailed_status_old
= None
3087 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
3089 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3090 self
._write
_op
_status
(nslcmop_id
, stage
)
3092 delete_timeout
= 20 * 60 # 20 minutes
3093 while delete_timeout
> 0:
3094 desc
= await self
.RO
.show(
3096 item_id_name
=ro_nsr_id
,
3097 extra_item
="action",
3098 extra_item_id
=ro_delete_action
)
3101 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3103 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3104 if ns_status
== "ERROR":
3105 raise ROclient
.ROClientException(ns_status_info
)
3106 elif ns_status
== "BUILD":
3107 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3108 elif ns_status
== "ACTIVE":
3109 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3110 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3113 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3114 if stage
[2] != detailed_status_old
:
3115 detailed_status_old
= stage
[2]
3116 db_nsr_update
["detailed-status"] = " ".join(stage
)
3117 self
._write
_op
_status
(nslcmop_id
, stage
)
3118 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3119 await asyncio
.sleep(5, loop
=self
.loop
)
3121 else: # delete_timeout <= 0:
3122 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
3124 except Exception as e
:
3125 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3126 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3127 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3128 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3129 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3130 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
3131 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3132 failed_detail
.append("delete conflict: {}".format(e
))
3133 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
3135 failed_detail
.append("delete error: {}".format(e
))
3136 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
3139 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
3140 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
3142 stage
[2] = "Deleting nsd from RO."
3143 db_nsr_update
["detailed-status"] = " ".join(stage
)
3144 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3145 self
._write
_op
_status
(nslcmop_id
, stage
)
3146 await self
.RO
.delete("nsd", ro_nsd_id
)
3147 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
3148 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3149 except Exception as e
:
3150 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3151 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3152 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
3153 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3154 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
3155 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3157 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
3158 self
.logger
.error(logging_text
+ failed_detail
[-1])
3160 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
3161 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
3162 if not vnf_deployed
or not vnf_deployed
["id"]:
3165 ro_vnfd_id
= vnf_deployed
["id"]
3166 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
3167 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
3168 db_nsr_update
["detailed-status"] = " ".join(stage
)
3169 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3170 self
._write
_op
_status
(nslcmop_id
, stage
)
3171 await self
.RO
.delete("vnfd", ro_vnfd_id
)
3172 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
3173 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3174 except Exception as e
:
3175 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3176 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3177 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
3178 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3179 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
3180 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3182 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
3183 self
.logger
.error(logging_text
+ failed_detail
[-1])
3186 stage
[2] = "Error deleting from VIM"
3188 stage
[2] = "Deleted from VIM"
3189 db_nsr_update
["detailed-status"] = " ".join(stage
)
3190 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3191 self
._write
_op
_status
(nslcmop_id
, stage
)
3194 raise LcmException("; ".join(failed_detail
))
3196 async def terminate(self
, nsr_id
, nslcmop_id
):
3197 # Try to lock HA task here
3198 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3199 if not task_is_locked_by_me
:
3202 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
3203 self
.logger
.debug(logging_text
+ "Enter")
3204 timeout_ns_terminate
= self
.timeout_ns_terminate
3207 operation_params
= None
3209 error_list
= [] # annotates all failed error messages
3210 db_nslcmop_update
= {}
3211 autoremove
= False # autoremove after terminated
3212 tasks_dict_info
= {}
3214 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
3215 # ^ contains [stage, step, VIM-status]
3217 # wait for any previous tasks in process
3218 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
3220 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
3221 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3222 operation_params
= db_nslcmop
.get("operationParams") or {}
3223 if operation_params
.get("timeout_ns_terminate"):
3224 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
3225 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
3226 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3228 db_nsr_update
["operational-status"] = "terminating"
3229 db_nsr_update
["config-status"] = "terminating"
3230 self
._write
_ns
_status
(
3232 ns_state
="TERMINATING",
3233 current_operation
="TERMINATING",
3234 current_operation_id
=nslcmop_id
,
3235 other_update
=db_nsr_update
3237 self
._write
_op
_status
(
3242 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
3243 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
3246 stage
[1] = "Getting vnf descriptors from db."
3247 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
3248 db_vnfds_from_id
= {}
3249 db_vnfds_from_member_index
= {}
3251 for vnfr
in db_vnfrs_list
:
3252 vnfd_id
= vnfr
["vnfd-id"]
3253 if vnfd_id
not in db_vnfds_from_id
:
3254 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
3255 db_vnfds_from_id
[vnfd_id
] = vnfd
3256 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
3258 # Destroy individual execution environments when there are terminating primitives.
3259 # Rest of EE will be deleted at once
3260 # TODO - check before calling _destroy_N2VC
3261 # if not operation_params.get("skip_terminate_primitives"):#
3262 # or not vca.get("needed_terminate"):
3263 stage
[0] = "Stage 2/3 execute terminating primitives."
3264 self
.logger
.debug(logging_text
+ stage
[0])
3265 stage
[1] = "Looking execution environment that needs terminate."
3266 self
.logger
.debug(logging_text
+ stage
[1])
3267 # self.logger.debug("nsr_deployed: {}".format(nsr_deployed))
3268 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
3269 config_descriptor
= None
3270 if not vca
or not vca
.get("ee_id"):
3272 if not vca
.get("member-vnf-index"):
3274 config_descriptor
= db_nsr
.get("ns-configuration")
3275 elif vca
.get("vdu_id"):
3276 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3277 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
3279 config_descriptor
= vdud
.get("vdu-configuration")
3280 elif vca
.get("kdu_name"):
3281 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3282 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
3284 config_descriptor
= kdud
.get("kdu-configuration")
3286 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
3287 vca_type
= vca
.get("type")
3288 exec_terminate_primitives
= (not operation_params
.get("skip_terminate_primitives") and
3289 vca
.get("needed_terminate"))
3290 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
3291 # pending native charms
3292 destroy_ee
= True if vca_type
in ("helm", "native_charm") else False
3293 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
3294 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
3295 task
= asyncio
.ensure_future(
3296 self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
, vca_index
,
3297 destroy_ee
, exec_terminate_primitives
))
3298 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
3300 # wait for pending tasks of terminate primitives
3302 self
.logger
.debug(logging_text
+ 'Waiting for tasks {}'.format(list(tasks_dict_info
.keys())))
3303 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
3304 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
3306 tasks_dict_info
.clear()
3308 return # raise LcmException("; ".join(error_list))
3310 # remove All execution environments at once
3311 stage
[0] = "Stage 3/3 delete all."
3313 if nsr_deployed
.get("VCA"):
3314 stage
[1] = "Deleting all execution environments."
3315 self
.logger
.debug(logging_text
+ stage
[1])
3316 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
3317 timeout
=self
.timeout_charm_delete
))
3318 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
3319 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
3321 # Delete from k8scluster
3322 stage
[1] = "Deleting KDUs."
3323 self
.logger
.debug(logging_text
+ stage
[1])
3324 # print(nsr_deployed)
3325 for kdu
in get_iterable(nsr_deployed
, "K8s"):
3326 if not kdu
or not kdu
.get("kdu-instance"):
3328 kdu_instance
= kdu
.get("kdu-instance")
3329 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
3330 task_delete_kdu_instance
= asyncio
.ensure_future(
3331 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
3332 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3333 kdu_instance
=kdu_instance
))
3335 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
3336 format(kdu
.get("k8scluster-type")))
3338 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
3341 stage
[1] = "Deleting ns from VIM."
3343 task_delete_ro
= asyncio
.ensure_future(
3344 self
._terminate
_ng
_ro
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3346 task_delete_ro
= asyncio
.ensure_future(
3347 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3348 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
3350 # rest of staff will be done at finally
3352 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
3353 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3355 except asyncio
.CancelledError
:
3356 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
3357 exc
= "Operation was cancelled"
3358 except Exception as e
:
3359 exc
= traceback
.format_exc()
3360 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
3363 error_list
.append(str(exc
))
3365 # wait for pending tasks
3367 stage
[1] = "Waiting for terminate pending tasks."
3368 self
.logger
.debug(logging_text
+ stage
[1])
3369 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
3371 stage
[1] = stage
[2] = ""
3372 except asyncio
.CancelledError
:
3373 error_list
.append("Cancelled")
3374 # TODO cancell all tasks
3375 except Exception as exc
:
3376 error_list
.append(str(exc
))
3377 # update status at database
3379 error_detail
= "; ".join(error_list
)
3380 # self.logger.error(logging_text + error_detail)
3381 error_description_nslcmop
= '{} Detail: {}'.format(stage
[0], error_detail
)
3382 error_description_nsr
= 'Operation: TERMINATING.{}, {}.'.format(nslcmop_id
, stage
[0])
3384 db_nsr_update
["operational-status"] = "failed"
3385 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
3386 db_nslcmop_update
["detailed-status"] = error_detail
3387 nslcmop_operation_state
= "FAILED"
3391 error_description_nsr
= error_description_nslcmop
= None
3392 ns_state
= "NOT_INSTANTIATED"
3393 db_nsr_update
["operational-status"] = "terminated"
3394 db_nsr_update
["detailed-status"] = "Done"
3395 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
3396 db_nslcmop_update
["detailed-status"] = "Done"
3397 nslcmop_operation_state
= "COMPLETED"
3400 self
._write
_ns
_status
(
3403 current_operation
="IDLE",
3404 current_operation_id
=None,
3405 error_description
=error_description_nsr
,
3406 error_detail
=error_detail
,
3407 other_update
=db_nsr_update
3409 self
._write
_op
_status
(
3412 error_message
=error_description_nslcmop
,
3413 operation_state
=nslcmop_operation_state
,
3414 other_update
=db_nslcmop_update
,
3416 if ns_state
== "NOT_INSTANTIATED":
3418 self
.db
.set_list("vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "NOT_INSTANTIATED"})
3419 except DbException
as e
:
3420 self
.logger
.warn(logging_text
+ 'Error writing VNFR status for nsr-id-ref: {} -> {}'.
3422 if operation_params
:
3423 autoremove
= operation_params
.get("autoremove", False)
3424 if nslcmop_operation_state
:
3426 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3427 "operationState": nslcmop_operation_state
,
3428 "autoremove": autoremove
},
3430 except Exception as e
:
3431 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3433 self
.logger
.debug(logging_text
+ "Exit")
3434 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
3436 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
3438 error_detail_list
= []
3440 pending_tasks
= list(created_tasks_info
.keys())
3441 num_tasks
= len(pending_tasks
)
3443 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3444 self
._write
_op
_status
(nslcmop_id
, stage
)
3445 while pending_tasks
:
3447 _timeout
= timeout
+ time_start
- time()
3448 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
3449 return_when
=asyncio
.FIRST_COMPLETED
)
3450 num_done
+= len(done
)
3451 if not done
: # Timeout
3452 for task
in pending_tasks
:
3453 new_error
= created_tasks_info
[task
] + ": Timeout"
3454 error_detail_list
.append(new_error
)
3455 error_list
.append(new_error
)
3458 if task
.cancelled():
3461 exc
= task
.exception()
3463 if isinstance(exc
, asyncio
.TimeoutError
):
3465 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
3466 error_list
.append(created_tasks_info
[task
])
3467 error_detail_list
.append(new_error
)
3468 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
3470 self
.logger
.error(logging_text
+ new_error
)
3472 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
3473 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
3475 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
3476 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3478 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
3479 if nsr_id
: # update also nsr
3480 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
3481 "errorDetail": ". ".join(error_detail_list
)})
3482 self
._write
_op
_status
(nslcmop_id
, stage
)
3483 return error_detail_list
3486 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
3488 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
3489 The default-value is used. If it is between < > it look for a value at instantiation_params
3490 :param primitive_desc: portion of VNFD/NSD that describes primitive
3491 :param params: Params provided by user
3492 :param instantiation_params: Instantiation params provided by user
3493 :return: a dictionary with the calculated params
3495 calculated_params
= {}
3496 for parameter
in primitive_desc
.get("parameter", ()):
3497 param_name
= parameter
["name"]
3498 if param_name
in params
:
3499 calculated_params
[param_name
] = params
[param_name
]
3500 elif "default-value" in parameter
or "value" in parameter
:
3501 if "value" in parameter
:
3502 calculated_params
[param_name
] = parameter
["value"]
3504 calculated_params
[param_name
] = parameter
["default-value"]
3505 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3506 and calculated_params
[param_name
].endswith(">"):
3507 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3508 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3510 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3511 format(calculated_params
[param_name
], primitive_desc
["name"]))
3513 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3514 format(param_name
, primitive_desc
["name"]))
3516 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3517 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3519 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3520 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3522 # add always ns_config_info if primitive name is config
3523 if primitive_desc
["name"] == "config":
3524 if "ns_config_info" in instantiation_params
:
3525 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3526 return calculated_params
3528 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None,
3529 ee_descriptor_id
=None):
3530 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3531 for vca
in deployed_vca
:
3534 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3536 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3538 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3540 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
3544 # vca_deployed not found
3545 raise LcmException("charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
3546 " is not deployed".format(member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
,
3550 ee_id
= vca
.get("ee_id")
3551 vca_type
= vca
.get("type", "lxc_proxy_charm") # default value for backward compatibility - proxy charm
3553 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3554 "execution environment"
3555 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3556 return ee_id
, vca_type
3558 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3559 retries_interval
=30, timeout
=None,
3560 vca_type
=None, db_dict
=None) -> (str, str):
3562 if primitive
== "config":
3563 primitive_params
= {"params": primitive_params
}
3565 vca_type
= vca_type
or "lxc_proxy_charm"
3569 output
= await asyncio
.wait_for(
3570 self
.vca_map
[vca_type
].exec_primitive(
3572 primitive_name
=primitive
,
3573 params_dict
=primitive_params
,
3574 progress_timeout
=self
.timeout_progress_primitive
,
3575 total_timeout
=self
.timeout_primitive
,
3577 timeout
=timeout
or self
.timeout_primitive
)
3580 except asyncio
.CancelledError
:
3582 except Exception as e
: # asyncio.TimeoutError
3583 if isinstance(e
, asyncio
.TimeoutError
):
3587 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3589 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3591 return 'FAILED', str(e
)
3593 return 'COMPLETED', output
3595 except (LcmException
, asyncio
.CancelledError
):
3597 except Exception as e
:
3598 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3600 async def action(self
, nsr_id
, nslcmop_id
):
3602 # Try to lock HA task here
3603 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3604 if not task_is_locked_by_me
:
3607 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3608 self
.logger
.debug(logging_text
+ "Enter")
3609 # get all needed from database
3613 db_nslcmop_update
= {}
3614 nslcmop_operation_state
= None
3615 error_description_nslcmop
= None
3618 # wait for any previous tasks in process
3619 step
= "Waiting for previous operations to terminate"
3620 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3622 self
._write
_ns
_status
(
3625 current_operation
="RUNNING ACTION",
3626 current_operation_id
=nslcmop_id
3629 step
= "Getting information from database"
3630 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3631 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3633 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3634 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3635 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3636 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3637 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3638 primitive
= db_nslcmop
["operationParams"]["primitive"]
3639 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3640 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3643 step
= "Getting vnfr from database"
3644 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3645 step
= "Getting vnfd from database"
3646 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3648 step
= "Getting nsd from database"
3649 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3651 # for backward compatibility
3652 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3653 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3654 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3655 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3657 # look for primitive
3658 config_primitive_desc
= descriptor_configuration
= None
3660 for vdu
in get_iterable(db_vnfd
, "vdu"):
3661 if vdu_id
== vdu
["id"]:
3662 descriptor_configuration
= vdu
.get("vdu-configuration")
3665 for kdu
in get_iterable(db_vnfd
, "kdu"):
3666 if kdu_name
== kdu
["name"]:
3667 descriptor_configuration
= kdu
.get("kdu-configuration")
3670 descriptor_configuration
= db_vnfd
.get("vnf-configuration")
3672 descriptor_configuration
= db_nsd
.get("ns-configuration")
3674 if descriptor_configuration
and descriptor_configuration
.get("config-primitive"):
3675 for config_primitive
in descriptor_configuration
["config-primitive"]:
3676 if config_primitive
["name"] == primitive
:
3677 config_primitive_desc
= config_primitive
3680 if not config_primitive_desc
:
3681 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3682 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3684 primitive_name
= primitive
3685 ee_descriptor_id
= None
3687 primitive_name
= config_primitive_desc
.get("execution-environment-primitive", primitive
)
3688 ee_descriptor_id
= config_primitive_desc
.get("execution-environment-ref")
3692 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3693 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3695 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3696 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3698 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3700 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3703 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3705 # TODO check if ns is in a proper status
3706 if kdu_name
and (primitive_name
in ("upgrade", "rollback", "status") or kdu_action
):
3707 # kdur and desc_params already set from before
3708 if primitive_params
:
3709 desc_params
.update(primitive_params
)
3710 # TODO Check if we will need something at vnf level
3711 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3712 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3715 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3717 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3718 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3719 raise LcmException(msg
)
3721 db_dict
= {"collection": "nsrs",
3722 "filter": {"_id": nsr_id
},
3723 "path": "_admin.deployed.K8s.{}".format(index
)}
3724 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
))
3725 step
= "Executing kdu {}".format(primitive_name
)
3726 if primitive_name
== "upgrade":
3727 if desc_params
.get("kdu_model"):
3728 kdu_model
= desc_params
.get("kdu_model")
3729 del desc_params
["kdu_model"]
3731 kdu_model
= kdu
.get("kdu-model")
3732 parts
= kdu_model
.split(sep
=":")
3734 kdu_model
= parts
[0]
3736 detailed_status
= await asyncio
.wait_for(
3737 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3738 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3739 kdu_instance
=kdu
.get("kdu-instance"),
3740 atomic
=True, kdu_model
=kdu_model
,
3741 params
=desc_params
, db_dict
=db_dict
,
3742 timeout
=timeout_ns_action
),
3743 timeout
=timeout_ns_action
+ 10)
3744 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3745 elif primitive_name
== "rollback":
3746 detailed_status
= await asyncio
.wait_for(
3747 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3748 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3749 kdu_instance
=kdu
.get("kdu-instance"),
3751 timeout
=timeout_ns_action
)
3752 elif primitive_name
== "status":
3753 detailed_status
= await asyncio
.wait_for(
3754 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3755 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3756 kdu_instance
=kdu
.get("kdu-instance")),
3757 timeout
=timeout_ns_action
)
3759 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3760 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3762 detailed_status
= await asyncio
.wait_for(
3763 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3764 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3765 kdu_instance
=kdu_instance
,
3766 primitive_name
=primitive_name
,
3767 params
=params
, db_dict
=db_dict
,
3768 timeout
=timeout_ns_action
),
3769 timeout
=timeout_ns_action
)
3772 nslcmop_operation_state
= 'COMPLETED'
3774 detailed_status
= ''
3775 nslcmop_operation_state
= 'FAILED'
3777 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3778 member_vnf_index
=vnf_index
,
3780 vdu_count_index
=vdu_count_index
,
3781 ee_descriptor_id
=ee_descriptor_id
)
3782 db_nslcmop_notif
= {"collection": "nslcmops",
3783 "filter": {"_id": nslcmop_id
},
3784 "path": "admin.VCA"}
3785 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3787 primitive
=primitive_name
,
3788 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3789 timeout
=timeout_ns_action
,
3791 db_dict
=db_nslcmop_notif
)
3793 db_nslcmop_update
["detailed-status"] = detailed_status
3794 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3795 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3797 return # database update is called inside finally
3799 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3800 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3802 except asyncio
.CancelledError
:
3803 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3804 exc
= "Operation was cancelled"
3805 except asyncio
.TimeoutError
:
3806 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3808 except Exception as e
:
3809 exc
= traceback
.format_exc()
3810 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3813 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3814 "FAILED {}: {}".format(step
, exc
)
3815 nslcmop_operation_state
= "FAILED"
3817 self
._write
_ns
_status
(
3819 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3820 current_operation
="IDLE",
3821 current_operation_id
=None,
3822 # error_description=error_description_nsr,
3823 # error_detail=error_detail,
3824 other_update
=db_nsr_update
3827 self
._write
_op
_status
(
3830 error_message
=error_description_nslcmop
,
3831 operation_state
=nslcmop_operation_state
,
3832 other_update
=db_nslcmop_update
,
3835 if nslcmop_operation_state
:
3837 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3838 "operationState": nslcmop_operation_state
},
3840 except Exception as e
:
3841 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3842 self
.logger
.debug(logging_text
+ "Exit")
3843 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3844 return nslcmop_operation_state
, detailed_status
3846 async def scale(self
, nsr_id
, nslcmop_id
):
3848 # Try to lock HA task here
3849 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3850 if not task_is_locked_by_me
:
3853 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3854 self
.logger
.debug(logging_text
+ "Enter")
3855 # get all needed from database
3858 db_nslcmop_update
= {}
3859 nslcmop_operation_state
= None
3862 # in case of error, indicates what part of scale was failed to put nsr at error status
3863 scale_process
= None
3864 old_operational_status
= ""
3865 old_config_status
= ""
3867 # wait for any previous tasks in process
3868 step
= "Waiting for previous operations to terminate"
3869 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3871 self
._write
_ns
_status
(
3874 current_operation
="SCALING",
3875 current_operation_id
=nslcmop_id
3878 step
= "Getting nslcmop from database"
3879 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3880 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3881 step
= "Getting nsr from database"
3882 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3884 old_operational_status
= db_nsr
["operational-status"]
3885 old_config_status
= db_nsr
["config-status"]
3886 step
= "Parsing scaling parameters"
3887 # self.logger.debug(step)
3888 db_nsr_update
["operational-status"] = "scaling"
3889 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3890 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3893 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3894 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3895 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3896 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3897 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3900 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3901 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3902 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3903 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3904 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3906 # for backward compatibility
3907 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3908 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3909 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3910 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3912 step
= "Getting vnfr from database"
3913 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3914 step
= "Getting vnfd from database"
3915 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3917 step
= "Getting scaling-group-descriptor"
3918 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3919 if scaling_descriptor
["name"] == scaling_group
:
3922 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3923 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3926 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3927 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3928 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3931 # TODO check if ns is in a proper status
3932 step
= "Sending scale order to VIM"
3934 if not db_nsr
["_admin"].get("scaling-group"):
3935 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3936 admin_scale_index
= 0
3938 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3939 if admin_scale_info
["name"] == scaling_group
:
3940 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3942 else: # not found, set index one plus last element and add new entry with the name
3943 admin_scale_index
+= 1
3944 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3945 RO_scaling_info
= []
3946 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3947 if scaling_type
== "SCALE_OUT":
3948 # count if max-instance-count is reached
3949 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3950 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3951 if nb_scale_op
>= max_instance_count
:
3952 raise LcmException("reached the limit of {} (max-instance-count) "
3953 "scaling-out operations for the "
3954 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3957 vdu_scaling_info
["scaling_direction"] = "OUT"
3958 vdu_scaling_info
["vdu-create"] = {}
3959 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3960 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3961 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3962 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3964 elif scaling_type
== "SCALE_IN":
3965 # count if min-instance-count is reached
3966 min_instance_count
= 0
3967 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3968 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3969 if nb_scale_op
<= min_instance_count
:
3970 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3971 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3973 vdu_scaling_info
["scaling_direction"] = "IN"
3974 vdu_scaling_info
["vdu-delete"] = {}
3975 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3976 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3977 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3978 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3980 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3981 vdu_create
= vdu_scaling_info
.get("vdu-create")
3982 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3983 if vdu_scaling_info
["scaling_direction"] == "IN":
3984 for vdur
in reversed(db_vnfr
["vdur"]):
3985 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3986 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3987 vdu_scaling_info
["vdu"].append({
3988 "name": vdur
["name"],
3989 "vdu_id": vdur
["vdu-id-ref"],
3992 for interface
in vdur
["interfaces"]:
3993 vdu_scaling_info
["vdu"][-1]["interface"].append({
3994 "name": interface
["name"],
3995 "ip_address": interface
["ip-address"],
3996 "mac_address": interface
.get("mac-address"),
3998 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
4001 step
= "Executing pre-scale vnf-config-primitive"
4002 if scaling_descriptor
.get("scaling-config-action"):
4003 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
4004 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
4005 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
4006 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
4007 step
= db_nslcmop_update
["detailed-status"] = \
4008 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
4010 # look for primitive
4011 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
4012 if config_primitive
["name"] == vnf_config_primitive
:
4016 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
4017 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
4018 "primitive".format(scaling_group
, vnf_config_primitive
))
4020 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
4021 if db_vnfr
.get("additionalParamsForVnf"):
4022 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
4024 scale_process
= "VCA"
4025 db_nsr_update
["config-status"] = "configuring pre-scaling"
4026 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
4028 # Pre-scale retry check: Check if this sub-operation has been executed before
4029 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4030 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
4031 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4032 # Skip sub-operation
4033 result
= 'COMPLETED'
4034 result_detail
= 'Done'
4035 self
.logger
.debug(logging_text
+
4036 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
4037 vnf_config_primitive
, result
, result_detail
))
4039 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4040 # New sub-operation: Get index of this sub-operation
4041 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4042 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
4043 format(vnf_config_primitive
))
4045 # retry: Get registered params for this existing sub-operation
4046 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4047 vnf_index
= op
.get('member_vnf_index')
4048 vnf_config_primitive
= op
.get('primitive')
4049 primitive_params
= op
.get('primitive_params')
4050 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
4051 format(vnf_config_primitive
))
4052 # Execute the primitive, either with new (first-time) or registered (reintent) args
4053 ee_descriptor_id
= config_primitive
.get("execution-environment-ref")
4054 primitive_name
= config_primitive
.get("execution-environment-primitive",
4055 vnf_config_primitive
)
4056 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
4057 member_vnf_index
=vnf_index
,
4059 vdu_count_index
=None,
4060 ee_descriptor_id
=ee_descriptor_id
)
4061 result
, result_detail
= await self
._ns
_execute
_primitive
(
4062 ee_id
, primitive_name
, primitive_params
, vca_type
)
4063 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
4064 vnf_config_primitive
, result
, result_detail
))
4065 # Update operationState = COMPLETED | FAILED
4066 self
._update
_suboperation
_status
(
4067 db_nslcmop
, op_index
, result
, result_detail
)
4069 if result
== "FAILED":
4070 raise LcmException(result_detail
)
4071 db_nsr_update
["config-status"] = old_config_status
4072 scale_process
= None
4076 # Should this block be skipped if 'RO_nsr_id' == None ?
4077 # if (RO_nsr_id and RO_scaling_info):
4079 scale_process
= "RO"
4080 # Scale RO retry check: Check if this sub-operation has been executed before
4081 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4082 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
4083 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4084 # Skip sub-operation
4085 result
= 'COMPLETED'
4086 result_detail
= 'Done'
4087 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
4088 result
, result_detail
))
4090 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4091 # New sub-operation: Get index of this sub-operation
4092 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4093 self
.logger
.debug(logging_text
+ "New sub-operation RO")
4095 # retry: Get registered params for this existing sub-operation
4096 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4097 RO_nsr_id
= op
.get('RO_nsr_id')
4098 RO_scaling_info
= op
.get('RO_scaling_info')
4099 self
.logger
.debug(logging_text
+ "Sub-operation RO retry for primitive {}".format(
4100 vnf_config_primitive
))
4102 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
4103 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
4104 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
4106 RO_nslcmop_id
= RO_desc
["instance_action_id"]
4107 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
4109 RO_task_done
= False
4110 step
= detailed_status
= "Waiting for VIM to scale. RO_task_id={}.".format(RO_nslcmop_id
)
4111 detailed_status_old
= None
4112 self
.logger
.debug(logging_text
+ step
)
4114 deployment_timeout
= 1 * 3600 # One hour
4115 while deployment_timeout
> 0:
4116 if not RO_task_done
:
4117 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
4118 extra_item_id
=RO_nslcmop_id
)
4121 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4123 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4124 if ns_status
== "ERROR":
4125 raise ROclient
.ROClientException(ns_status_info
)
4126 elif ns_status
== "BUILD":
4127 detailed_status
= step
+ "; {}".format(ns_status_info
)
4128 elif ns_status
== "ACTIVE":
4130 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
4131 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
4132 self
.logger
.debug(logging_text
+ step
)
4134 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
4136 desc
= await self
.RO
.show("ns", RO_nsr_id
)
4137 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
4139 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4141 if ns_status
== "ERROR":
4142 raise ROclient
.ROClientException(ns_status_info
)
4143 elif ns_status
== "BUILD":
4144 detailed_status
= step
+ "; {}".format(ns_status_info
)
4145 elif ns_status
== "ACTIVE":
4146 step
= detailed_status
= \
4147 "Waiting for management IP address reported by the VIM. Updating VNFRs"
4149 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
4150 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
4152 except LcmExceptionNoMgmtIP
:
4155 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
4156 if detailed_status
!= detailed_status_old
:
4157 self
._update
_suboperation
_status
(
4158 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
4159 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
4160 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
4162 await asyncio
.sleep(5, loop
=self
.loop
)
4163 deployment_timeout
-= 5
4164 if deployment_timeout
<= 0:
4165 self
._update
_suboperation
_status
(
4166 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
4167 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
4169 # update VDU_SCALING_INFO with the obtained ip_addresses
4170 if vdu_scaling_info
["scaling_direction"] == "OUT":
4171 for vdur
in reversed(db_vnfr
["vdur"]):
4172 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
4173 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
4174 vdu_scaling_info
["vdu"].append({
4175 "name": vdur
["name"],
4176 "vdu_id": vdur
["vdu-id-ref"],
4179 for interface
in vdur
["interfaces"]:
4180 vdu_scaling_info
["vdu"][-1]["interface"].append({
4181 "name": interface
["name"],
4182 "ip_address": interface
["ip-address"],
4183 "mac_address": interface
.get("mac-address"),
4185 del vdu_scaling_info
["vdu-create"]
4187 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
4190 scale_process
= None
4192 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4195 # execute primitive service POST-SCALING
4196 step
= "Executing post-scale vnf-config-primitive"
4197 if scaling_descriptor
.get("scaling-config-action"):
4198 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
4199 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
4200 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
4201 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
4202 step
= db_nslcmop_update
["detailed-status"] = \
4203 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
4205 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
4206 if db_vnfr
.get("additionalParamsForVnf"):
4207 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
4209 # look for primitive
4210 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
4211 if config_primitive
["name"] == vnf_config_primitive
:
4215 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
4216 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
4217 "config-primitive".format(scaling_group
, vnf_config_primitive
))
4218 scale_process
= "VCA"
4219 db_nsr_update
["config-status"] = "configuring post-scaling"
4220 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
4222 # Post-scale retry check: Check if this sub-operation has been executed before
4223 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4224 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
4225 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4226 # Skip sub-operation
4227 result
= 'COMPLETED'
4228 result_detail
= 'Done'
4229 self
.logger
.debug(logging_text
+
4230 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
4231 format(vnf_config_primitive
, result
, result_detail
))
4233 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4234 # New sub-operation: Get index of this sub-operation
4235 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4236 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
4237 format(vnf_config_primitive
))
4239 # retry: Get registered params for this existing sub-operation
4240 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4241 vnf_index
= op
.get('member_vnf_index')
4242 vnf_config_primitive
= op
.get('primitive')
4243 primitive_params
= op
.get('primitive_params')
4244 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
4245 format(vnf_config_primitive
))
4246 # Execute the primitive, either with new (first-time) or registered (reintent) args
4247 ee_descriptor_id
= config_primitive
.get("execution-environment-ref")
4248 primitive_name
= config_primitive
.get("execution-environment-primitive",
4249 vnf_config_primitive
)
4250 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
4251 member_vnf_index
=vnf_index
,
4253 vdu_count_index
=None,
4254 ee_descriptor_id
=ee_descriptor_id
)
4255 result
, result_detail
= await self
._ns
_execute
_primitive
(
4256 ee_id
, primitive_name
, primitive_params
, vca_type
)
4257 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
4258 vnf_config_primitive
, result
, result_detail
))
4259 # Update operationState = COMPLETED | FAILED
4260 self
._update
_suboperation
_status
(
4261 db_nslcmop
, op_index
, result
, result_detail
)
4263 if result
== "FAILED":
4264 raise LcmException(result_detail
)
4265 db_nsr_update
["config-status"] = old_config_status
4266 scale_process
= None
4269 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
4270 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
4271 else old_operational_status
4272 db_nsr_update
["config-status"] = old_config_status
4274 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
4275 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4277 except asyncio
.CancelledError
:
4278 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
4279 exc
= "Operation was cancelled"
4280 except Exception as e
:
4281 exc
= traceback
.format_exc()
4282 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
4284 self
._write
_ns
_status
(
4287 current_operation
="IDLE",
4288 current_operation_id
=None
4291 db_nslcmop_update
["detailed-status"] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
4292 nslcmop_operation_state
= "FAILED"
4294 db_nsr_update
["operational-status"] = old_operational_status
4295 db_nsr_update
["config-status"] = old_config_status
4296 db_nsr_update
["detailed-status"] = ""
4298 if "VCA" in scale_process
:
4299 db_nsr_update
["config-status"] = "failed"
4300 if "RO" in scale_process
:
4301 db_nsr_update
["operational-status"] = "failed"
4302 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
4305 error_description_nslcmop
= None
4306 nslcmop_operation_state
= "COMPLETED"
4307 db_nslcmop_update
["detailed-status"] = "Done"
4309 self
._write
_op
_status
(
4312 error_message
=error_description_nslcmop
,
4313 operation_state
=nslcmop_operation_state
,
4314 other_update
=db_nslcmop_update
,
4317 self
._write
_ns
_status
(
4320 current_operation
="IDLE",
4321 current_operation_id
=None,
4322 other_update
=db_nsr_update
4325 if nslcmop_operation_state
:
4327 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
4328 "operationState": nslcmop_operation_state
},
4331 # await asyncio.sleep(cooldown_time, loop=self.loop)
4332 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
4333 except Exception as e
:
4334 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
4335 self
.logger
.debug(logging_text
+ "Exit")
4336 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
4338 async def add_prometheus_metrics(self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
):
4339 if not self
.prometheus
:
4341 # look if exist a file called 'prometheus*.j2' and
4342 artifact_content
= self
.fs
.dir_ls(artifact_path
)
4343 job_file
= next((f
for f
in artifact_content
if f
.startswith("prometheus") and f
.endswith(".j2")), None)
4346 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
4350 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
4351 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
4353 vnfr_id
= vnfr_id
.replace("-", "")
4355 "JOB_NAME": vnfr_id
,
4356 "TARGET_IP": target_ip
,
4357 "EXPORTER_POD_IP": host_name
,
4358 "EXPORTER_POD_PORT": host_port
,
4360 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
4361 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
4362 for job
in job_list
:
4363 if not isinstance(job
.get("job_name"), str) or vnfr_id
not in job
["job_name"]:
4364 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
4365 job
["nsr_id"] = nsr_id
4366 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
4367 if await self
.prometheus
.update(job_dict
):
4368 return list(job_dict
.keys())