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 for index
in range(0, vdu_create
[vdu_id_ref
]):
633 vdur
= deepcopy(vdur
)
634 vdur
["_id"] = str(uuid4())
635 vdur
["count-index"] += 1
636 vdurs
.insert(vdu_index
+1+index
, vdur
)
637 del vdu_create
[vdu_id_ref
]
638 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
640 vdu_delete
[vdu_id_ref
] -= 1
641 if not vdu_delete
[vdu_id_ref
]:
642 del vdu_delete
[vdu_id_ref
]
643 # check all operations are done
644 if vdu_create
or vdu_delete
:
645 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
648 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
651 vnfr_update
= {"vdur": vdurs
}
652 db_vnfr
["vdur"] = vdurs
653 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
655 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
657 Updates database nsr with the RO info for the created vld
658 :param ns_update_nsr: dictionary to be filled with the updated info
659 :param db_nsr: content of db_nsr. This is also modified
660 :param nsr_desc_RO: nsr descriptor from RO
661 :return: Nothing, LcmException is raised on errors
664 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
665 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
666 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
668 vld
["vim-id"] = net_RO
.get("vim_net_id")
669 vld
["name"] = net_RO
.get("vim_name")
670 vld
["status"] = net_RO
.get("status")
671 vld
["status-detailed"] = net_RO
.get("error_msg")
672 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
675 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
677 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
679 for db_vnfr
in db_vnfrs
.values():
680 vnfr_update
= {"status": "ERROR"}
681 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
682 if "status" not in vdur
:
683 vdur
["status"] = "ERROR"
684 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
686 vdur
["status-detailed"] = str(error_text
)
687 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
688 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
689 except DbException
as e
:
690 self
.logger
.error("Cannot update vnf. {}".format(e
))
692 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
694 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
695 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
696 :param nsr_desc_RO: nsr descriptor from RO
697 :return: Nothing, LcmException is raised on errors
699 for vnf_index
, db_vnfr
in db_vnfrs
.items():
700 for vnf_RO
in nsr_desc_RO
["vnfs"]:
701 if vnf_RO
["member_vnf_index"] != vnf_index
:
704 if vnf_RO
.get("ip_address"):
705 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
706 elif not db_vnfr
.get("ip-address"):
707 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
708 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
710 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
711 vdur_RO_count_index
= 0
712 if vdur
.get("pdu-type"):
714 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
715 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
717 if vdur
["count-index"] != vdur_RO_count_index
:
718 vdur_RO_count_index
+= 1
720 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
721 if vdur_RO
.get("ip_address"):
722 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
724 vdur
["ip-address"] = None
725 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
726 vdur
["name"] = vdur_RO
.get("vim_name")
727 vdur
["status"] = vdur_RO
.get("status")
728 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
729 for ifacer
in get_iterable(vdur
, "interfaces"):
730 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
731 if ifacer
["name"] == interface_RO
.get("internal_name"):
732 ifacer
["ip-address"] = interface_RO
.get("ip_address")
733 ifacer
["mac-address"] = interface_RO
.get("mac_address")
736 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
738 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
739 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
742 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
743 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
745 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
746 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
747 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
749 vld
["vim-id"] = net_RO
.get("vim_net_id")
750 vld
["name"] = net_RO
.get("vim_name")
751 vld
["status"] = net_RO
.get("status")
752 vld
["status-detailed"] = net_RO
.get("error_msg")
753 vnfr_update
["vld.{}".format(vld_index
)] = vld
756 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
757 vnf_index
, vld
["id"]))
759 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
763 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
765 def _get_ns_config_info(self
, nsr_id
):
767 Generates a mapping between vnf,vdu elements and the N2VC id
768 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
769 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
770 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
771 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
773 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
774 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
776 ns_config_info
= {"osm-config-mapping": mapping
}
777 for vca
in vca_deployed_list
:
778 if not vca
["member-vnf-index"]:
780 if not vca
["vdu_id"]:
781 mapping
[vca
["member-vnf-index"]] = vca
["application"]
783 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
785 return ns_config_info
788 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
, ee_descriptor_id
):
790 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
791 primitives as verify-ssh-credentials, or config when needed
792 :param desc_primitive_list: information of the descriptor
793 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
794 this element contains a ssh public key
795 :param ee_descriptor_id: execution environment descriptor id. It is the value of
796 XXX_configuration.execution-environment-list.INDEX.id; it can be None
797 :return: The modified list. Can ba an empty list, but always a list
800 primitive_list
= desc_primitive_list
or []
802 # filter primitives by ee_id
803 primitive_list
= [p
for p
in primitive_list
if p
.get("execution-environment-ref") == ee_descriptor_id
]
807 primitive_list
.sort(key
=lambda val
: int(val
['seq']))
809 # look for primitive config, and get the position. None if not present
810 config_position
= None
811 for index
, primitive
in enumerate(primitive_list
):
812 if primitive
["name"] == "config":
813 config_position
= index
816 # for NS, add always a config primitive if not present (bug 874)
817 if not vca_deployed
["member-vnf-index"] and config_position
is None:
818 primitive_list
.insert(0, {"name": "config", "parameter": []})
820 # TODO revise if needed: for VNF/VDU add verify-ssh-credentials after config
821 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
822 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
823 return primitive_list
825 async def _instantiate_ng_ro(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
826 n2vc_key_list
, stage
, start_deploy
, timeout_ns_deploy
):
827 nslcmop_id
= db_nslcmop
["_id"]
829 "name": db_nsr
["name"],
832 "image": deepcopy(db_nsr
["image"]),
833 "flavor": deepcopy(db_nsr
["flavor"]),
834 "action_id": nslcmop_id
,
836 for image
in target
["image"]:
837 image
["vim_info"] = []
838 for flavor
in target
["flavor"]:
839 flavor
["vim_info"] = []
841 ns_params
= db_nslcmop
.get("operationParams")
843 if ns_params
.get("ssh_keys"):
844 ssh_keys
+= ns_params
.get("ssh_keys")
846 ssh_keys
+= n2vc_key_list
849 for vld_index
, vld
in enumerate(nsd
.get("vld")):
850 target_vld
= {"id": vld
["id"],
852 "mgmt-network": vld
.get("mgmt-network", False),
853 "type": vld
.get("type"),
854 "vim_info": [{"vim-network-name": vld
.get("vim-network-name"),
855 "vim_account_id": ns_params
["vimAccountId"]}],
857 for cp
in vld
["vnfd-connection-point-ref"]:
858 cp2target
["member_vnf:{}.{}".format(cp
["member-vnf-index-ref"], cp
["vnfd-connection-point-ref"])] = \
859 "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
860 target
["ns"]["vld"].append(target_vld
)
861 for vnfr
in db_vnfrs
.values():
862 vnfd
= db_vnfds_ref
[vnfr
["vnfd-ref"]]
863 target_vnf
= deepcopy(vnfr
)
864 for vld
in target_vnf
.get("vld", ()):
865 # check if connected to a ns.vld
866 vnf_cp
= next((cp
for cp
in vnfd
.get("connection-point", ()) if
867 cp
.get("internal-vld-ref") == vld
["id"]), None)
869 ns_cp
= "member_vnf:{}.{}".format(vnfr
["member-vnf-index-ref"], vnf_cp
["id"])
870 if cp2target
.get(ns_cp
):
871 vld
["target"] = cp2target
[ns_cp
]
872 vld
["vim_info"] = [{"vim-network-name": vld
.get("vim-network-name"),
873 "vim_account_id": vnfr
["vim-account-id"]}]
875 for vdur
in target_vnf
.get("vdur", ()):
876 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
877 vdud_index
, vdud
= next(k
for k
in enumerate(vnfd
["vdu"]) if k
[1]["id"] == vdur
["vdu-id-ref"])
878 # vdur["additionalParams"] = vnfr.get("additionalParamsForVnf") # TODO additional params for VDU
881 if deep_get(vdud
, ("vdu-configuration", "config-access", "ssh-access", "required")):
882 vdur
["ssh-keys"] = ssh_keys
883 vdur
["ssh-access-required"] = True
884 elif deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required")) and \
885 any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"]):
886 vdur
["ssh-keys"] = ssh_keys
887 vdur
["ssh-access-required"] = True
890 if vdud
.get("cloud-init-file"):
891 vdur
["cloud-init"] = "{}:file:{}".format(vnfd
["_id"], vdud
.get("cloud-init-file"))
892 elif vdud
.get("cloud-init"):
893 vdur
["cloud-init"] = "{}:vdu:{}".format(vnfd
["_id"], vdud_index
)
896 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
897 if not next((vi
for vi
in ns_flavor
["vim_info"] if
898 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
899 ns_flavor
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
901 ns_image
= target
["image"][int(vdur
["ns-image-id"])]
902 if not next((vi
for vi
in ns_image
["vim_info"] if
903 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
904 ns_image
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
906 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
907 target
["vnf"].append(target_vnf
)
909 desc
= await self
.RO
.deploy(nsr_id
, target
)
910 action_id
= desc
["action_id"]
911 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
)
915 "_admin.deployed.RO.operational-status": "running",
916 "detailed-status": " ".join(stage
)
918 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
919 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
920 self
._write
_op
_status
(nslcmop_id
, stage
)
921 self
.logger
.debug(logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
))
924 async def _wait_ng_ro(self
, nsr_id
, action_id
, nslcmop_id
, start_time
, timeout
, stage
):
925 detailed_status_old
= None
927 while time() <= start_time
+ timeout
:
928 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
929 if desc_status
["status"] == "FAILED":
930 raise NgRoException(desc_status
["details"])
931 elif desc_status
["status"] == "BUILD":
932 stage
[2] = "VIM: ({})".format(desc_status
["details"])
933 elif desc_status
["status"] == "DONE":
934 stage
[2] = "Deployed at VIM"
937 assert False, "ROclient.check_ns_status returns unknown {}".format(desc_status
["status"])
938 if stage
[2] != detailed_status_old
:
939 detailed_status_old
= stage
[2]
940 db_nsr_update
["detailed-status"] = " ".join(stage
)
941 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
942 self
._write
_op
_status
(nslcmop_id
, stage
)
943 await asyncio
.sleep(5, loop
=self
.loop
)
944 else: # timeout_ns_deploy
945 raise NgRoException("Timeout waiting ns to deploy")
947 async def _terminate_ng_ro(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
951 start_deploy
= time()
959 desc
= await self
.RO
.deploy(nsr_id
, target
)
960 action_id
= desc
["action_id"]
961 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
962 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
963 self
.logger
.debug(logging_text
+ "ns terminate action at RO. action_id={}".format(action_id
))
966 delete_timeout
= 20 * 60 # 20 minutes
967 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
)
969 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
970 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
972 await self
.RO
.delete(nsr_id
)
973 except Exception as e
:
974 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
975 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
976 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
977 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
978 self
.logger
.debug(logging_text
+ "RO_action_id={} already deleted".format(action_id
))
979 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
980 failed_detail
.append("delete conflict: {}".format(e
))
981 self
.logger
.debug(logging_text
+ "RO_action_id={} delete conflict: {}".format(action_id
, e
))
983 failed_detail
.append("delete error: {}".format(e
))
984 self
.logger
.error(logging_text
+ "RO_action_id={} delete error: {}".format(action_id
, e
))
987 stage
[2] = "Error deleting from VIM"
989 stage
[2] = "Deleted from VIM"
990 db_nsr_update
["detailed-status"] = " ".join(stage
)
991 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
992 self
._write
_op
_status
(nslcmop_id
, stage
)
995 raise LcmException("; ".join(failed_detail
))
998 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
999 n2vc_key_list
, stage
):
1002 :param logging_text: preffix text to use at logging
1003 :param nsr_id: nsr identity
1004 :param nsd: database content of ns descriptor
1005 :param db_nsr: database content of ns record
1006 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
1008 :param db_vnfds_ref: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
1009 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
1010 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
1011 :return: None or exception
1015 RO_descriptor_number
= 0 # number of descriptors created at RO
1016 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
1017 nslcmop_id
= db_nslcmop
["_id"]
1018 start_deploy
= time()
1019 ns_params
= db_nslcmop
.get("operationParams")
1020 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1021 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1023 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1025 # Check for and optionally request placement optimization. Database will be updated if placement activated
1026 stage
[2] = "Waiting for Placement."
1027 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
1028 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
1029 for vnfr
in db_vnfrs
.values():
1030 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
1033 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
1036 return await self
._instantiate
_ng
_ro
(logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
,
1037 db_vnfds_ref
, n2vc_key_list
, stage
, start_deploy
,
1040 # get vnfds, instantiate at RO
1041 for c_vnf
in nsd
.get("constituent-vnfd", ()):
1042 member_vnf_index
= c_vnf
["member-vnf-index"]
1043 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
1044 vnfd_ref
= vnfd
["id"]
1046 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
1047 db_nsr_update
["detailed-status"] = " ".join(stage
)
1048 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1049 self
._write
_op
_status
(nslcmop_id
, stage
)
1051 # self.logger.debug(logging_text + stage[2])
1052 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
1053 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
1054 RO_descriptor_number
+= 1
1056 # look position at deployed.RO.vnfd if not present it will be appended at the end
1057 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
1058 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
1061 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
1062 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
1065 RO_update
= {"member-vnf-index": member_vnf_index
}
1066 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
1068 RO_update
["id"] = vnfd_list
[0]["uuid"]
1069 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
1070 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
1072 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
1073 get("additionalParamsForVnf"), nsr_id
)
1074 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
1075 RO_update
["id"] = desc
["uuid"]
1076 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
1077 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
1078 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
1079 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
1084 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1085 db_nsr_update
["detailed-status"] = " ".join(stage
)
1086 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1087 self
._write
_op
_status
(nslcmop_id
, stage
)
1089 # self.logger.debug(logging_text + stage[2])
1090 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
1091 RO_descriptor_number
+= 1
1092 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
1094 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
1095 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
1096 nsd_ref
, RO_nsd_uuid
))
1098 nsd_RO
= deepcopy(nsd
)
1099 nsd_RO
["id"] = RO_osm_nsd_id
1100 nsd_RO
.pop("_id", None)
1101 nsd_RO
.pop("_admin", None)
1102 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
1103 member_vnf_index
= c_vnf
["member-vnf-index"]
1104 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1105 for c_vld
in nsd_RO
.get("vld", ()):
1106 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
1107 member_vnf_index
= cp
["member-vnf-index-ref"]
1108 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1110 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
1111 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1112 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
1113 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
1114 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1117 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1118 db_nsr_update
["detailed-status"] = " ".join(stage
)
1119 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1120 self
._write
_op
_status
(nslcmop_id
, stage
)
1122 # if present use it unless in error status
1123 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
1126 stage
[2] = "Looking for existing ns at RO"
1127 db_nsr_update
["detailed-status"] = " ".join(stage
)
1128 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1129 self
._write
_op
_status
(nslcmop_id
, stage
)
1130 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
1131 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1133 except ROclient
.ROClientException
as e
:
1134 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
1136 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1138 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1139 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1140 if ns_status
== "ERROR":
1141 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
1142 self
.logger
.debug(logging_text
+ stage
[2])
1143 await self
.RO
.delete("ns", RO_nsr_id
)
1144 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1146 stage
[2] = "Checking dependencies"
1147 db_nsr_update
["detailed-status"] = " ".join(stage
)
1148 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1149 self
._write
_op
_status
(nslcmop_id
, stage
)
1150 # self.logger.debug(logging_text + stage[2])
1152 # check if VIM is creating and wait look if previous tasks in process
1153 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
1155 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
1156 self
.logger
.debug(logging_text
+ stage
[2])
1157 await asyncio
.wait(task_dependency
, timeout
=3600)
1158 if ns_params
.get("vnf"):
1159 for vnf
in ns_params
["vnf"]:
1160 if "vimAccountId" in vnf
:
1161 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
1162 vnf
["vimAccountId"])
1164 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
1165 self
.logger
.debug(logging_text
+ stage
[2])
1166 await asyncio
.wait(task_dependency
, timeout
=3600)
1168 stage
[2] = "Checking instantiation parameters."
1169 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, db_vnfrs
, n2vc_key_list
)
1170 stage
[2] = "Deploying ns at VIM."
1171 db_nsr_update
["detailed-status"] = " ".join(stage
)
1172 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1173 self
._write
_op
_status
(nslcmop_id
, stage
)
1175 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
1176 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
1177 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1178 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
1179 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
1181 # wait until NS is ready
1182 stage
[2] = "Waiting VIM to deploy ns."
1183 db_nsr_update
["detailed-status"] = " ".join(stage
)
1184 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1185 self
._write
_op
_status
(nslcmop_id
, stage
)
1186 detailed_status_old
= None
1187 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
1190 while time() <= start_deploy
+ timeout_ns_deploy
:
1191 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1194 if desc
!= old_desc
:
1195 # desc has changed => update db
1196 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
1199 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1200 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1201 if ns_status
== "ERROR":
1202 raise ROclient
.ROClientException(ns_status_info
)
1203 elif ns_status
== "BUILD":
1204 stage
[2] = "VIM: ({})".format(ns_status_info
)
1205 elif ns_status
== "ACTIVE":
1206 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
1208 self
.ns_update_vnfr(db_vnfrs
, desc
)
1210 except LcmExceptionNoMgmtIP
:
1213 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
1214 if stage
[2] != detailed_status_old
:
1215 detailed_status_old
= stage
[2]
1216 db_nsr_update
["detailed-status"] = " ".join(stage
)
1217 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1218 self
._write
_op
_status
(nslcmop_id
, stage
)
1219 await asyncio
.sleep(5, loop
=self
.loop
)
1220 else: # timeout_ns_deploy
1221 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
1224 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
1226 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
1227 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1228 stage
[2] = "Deployed at VIM"
1229 db_nsr_update
["detailed-status"] = " ".join(stage
)
1230 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1231 self
._write
_op
_status
(nslcmop_id
, stage
)
1232 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
1233 # self.logger.debug(logging_text + "Deployed at VIM")
1234 except (ROclient
.ROClientException
, LcmException
, DbException
, NgRoException
) as e
:
1235 stage
[2] = "ERROR deploying at VIM"
1236 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1239 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
1241 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1242 :param logging_text: prefix use for logging
1247 :param pub_key: public ssh key to inject, None to skip
1248 :param user: user to apply the public ssh key
1252 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
1256 target_vdu_id
= None
1262 if ro_retries
>= 360: # 1 hour
1263 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
1265 await asyncio
.sleep(10, loop
=self
.loop
)
1268 if not target_vdu_id
:
1269 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1271 if not vdu_id
: # for the VNF case
1272 if db_vnfr
.get("status") == "ERROR":
1273 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1274 ip_address
= db_vnfr
.get("ip-address")
1277 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1279 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1280 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1282 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1283 vdur
= db_vnfr
["vdur"][0]
1285 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1288 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1289 ip_address
= vdur
.get("ip-address")
1292 target_vdu_id
= vdur
["vdu-id-ref"]
1293 elif vdur
.get("status") == "ERROR":
1294 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1296 if not target_vdu_id
:
1299 # inject public key into machine
1300 if pub_key
and user
:
1301 # wait until NS is deployed at RO
1303 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1304 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1308 # self.logger.debug(logging_text + "Inserting RO key")
1309 if vdur
.get("pdu-type"):
1310 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1313 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1315 target
= {"action": "inject_ssh_key", "key": pub_key
, "user": user
,
1316 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdu_id
}]}],
1318 await self
.RO
.deploy(nsr_id
, target
)
1320 result_dict
= await self
.RO
.create_action(
1322 item_id_name
=ro_nsr_id
,
1323 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1325 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1326 if not result_dict
or not isinstance(result_dict
, dict):
1327 raise LcmException("Unknown response from RO when injecting key")
1328 for result
in result_dict
.values():
1329 if result
.get("vim_result") == 200:
1332 raise ROclient
.ROClientException("error injecting key: {}".format(
1333 result
.get("description")))
1335 except NgRoException
as e
:
1336 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1337 except ROclient
.ROClientException
as e
:
1339 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1343 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1349 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1351 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1353 my_vca
= vca_deployed_list
[vca_index
]
1354 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1355 # vdu or kdu: no dependencies
1359 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1360 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1361 configuration_status_list
= db_nsr
["configurationStatus"]
1362 for index
, vca_deployed
in enumerate(configuration_status_list
):
1363 if index
== vca_index
:
1366 if not my_vca
.get("member-vnf-index") or \
1367 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1368 internal_status
= configuration_status_list
[index
].get("status")
1369 if internal_status
== 'READY':
1371 elif internal_status
== 'BROKEN':
1372 raise LcmException("Configuration aborted because dependent charm/s has failed")
1376 # no dependencies, return
1378 await asyncio
.sleep(10)
1381 raise LcmException("Configuration aborted because dependent charm/s timeout")
1383 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1384 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
, vca_type
, vca_name
,
1385 ee_config_descriptor
):
1386 nsr_id
= db_nsr
["_id"]
1387 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1388 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1389 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1390 osm_config
= {"osm": {"ns_id": db_nsr
["_id"]}}
1392 'collection': 'nsrs',
1393 'filter': {'_id': nsr_id
},
1394 'path': db_update_entry
1400 element_under_configuration
= nsr_id
1404 vnfr_id
= db_vnfr
["_id"]
1405 osm_config
["osm"]["vnf_id"] = vnfr_id
1407 namespace
= "{nsi}.{ns}".format(
1408 nsi
=nsi_id
if nsi_id
else "",
1412 element_type
= 'VNF'
1413 element_under_configuration
= vnfr_id
1414 namespace
+= ".{}".format(vnfr_id
)
1416 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1417 element_type
= 'VDU'
1418 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1419 osm_config
["osm"]["vdu_id"] = vdu_id
1421 namespace
+= ".{}".format(kdu_name
)
1422 element_type
= 'KDU'
1423 element_under_configuration
= kdu_name
1424 osm_config
["osm"]["kdu_name"] = kdu_name
1427 artifact_path
= "{}/{}/{}/{}".format(
1428 base_folder
["folder"],
1429 base_folder
["pkg-dir"],
1430 "charms" if vca_type
in ("native_charm", "lxc_proxy_charm", "k8s_proxy_charm") else "helm-charts",
1433 # get initial_config_primitive_list that applies to this element
1434 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1436 # add config if not present for NS charm
1437 ee_descriptor_id
= ee_config_descriptor
.get("id")
1438 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1439 vca_deployed
, ee_descriptor_id
)
1441 # n2vc_redesign STEP 3.1
1442 # find old ee_id if exists
1443 ee_id
= vca_deployed
.get("ee_id")
1445 # create or register execution environment in VCA
1446 if vca_type
in ("lxc_proxy_charm", "k8s_proxy_charm", "helm"):
1448 self
._write
_configuration
_status
(
1450 vca_index
=vca_index
,
1452 element_under_configuration
=element_under_configuration
,
1453 element_type
=element_type
1456 step
= "create execution environment"
1457 self
.logger
.debug(logging_text
+ step
)
1458 ee_id
, credentials
= await self
.vca_map
[vca_type
].create_execution_environment(
1459 namespace
=namespace
,
1463 artifact_path
=artifact_path
,
1466 elif vca_type
== "native_charm":
1467 step
= "Waiting to VM being up and getting IP address"
1468 self
.logger
.debug(logging_text
+ step
)
1469 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1470 user
=None, pub_key
=None)
1471 credentials
= {"hostname": rw_mgmt_ip
}
1473 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1474 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1475 # merged. Meanwhile let's get username from initial-config-primitive
1476 if not username
and initial_config_primitive_list
:
1477 for config_primitive
in initial_config_primitive_list
:
1478 for param
in config_primitive
.get("parameter", ()):
1479 if param
["name"] == "ssh-username":
1480 username
= param
["value"]
1483 raise LcmException("Cannot determine the username neither with 'initial-config-primitive' nor with "
1484 "'config-access.ssh-access.default-user'")
1485 credentials
["username"] = username
1486 # n2vc_redesign STEP 3.2
1488 self
._write
_configuration
_status
(
1490 vca_index
=vca_index
,
1491 status
='REGISTERING',
1492 element_under_configuration
=element_under_configuration
,
1493 element_type
=element_type
1496 step
= "register execution environment {}".format(credentials
)
1497 self
.logger
.debug(logging_text
+ step
)
1498 ee_id
= await self
.vca_map
[vca_type
].register_execution_environment(
1499 credentials
=credentials
, namespace
=namespace
, db_dict
=db_dict
)
1501 # for compatibility with MON/POL modules, the need model and application name at database
1502 # TODO ask MON/POL if needed to not assuming anymore the format "model_name.application_name"
1503 ee_id_parts
= ee_id
.split('.')
1504 db_nsr_update
= {db_update_entry
+ "ee_id": ee_id
}
1505 if len(ee_id_parts
) >= 2:
1506 model_name
= ee_id_parts
[0]
1507 application_name
= ee_id_parts
[1]
1508 db_nsr_update
[db_update_entry
+ "model"] = model_name
1509 db_nsr_update
[db_update_entry
+ "application"] = application_name
1511 # n2vc_redesign STEP 3.3
1512 step
= "Install configuration Software"
1514 self
._write
_configuration
_status
(
1516 vca_index
=vca_index
,
1517 status
='INSTALLING SW',
1518 element_under_configuration
=element_under_configuration
,
1519 element_type
=element_type
,
1520 other_update
=db_nsr_update
1523 # TODO check if already done
1524 self
.logger
.debug(logging_text
+ step
)
1526 if vca_type
== "native_charm":
1527 config_primitive
= next((p
for p
in initial_config_primitive_list
if p
["name"] == "config"), None)
1528 if config_primitive
:
1529 config
= self
._map
_primitive
_params
(
1535 if vca_type
== "lxc_proxy_charm":
1536 if element_type
== "NS":
1537 num_units
= db_nsr
.get("config-units") or 1
1538 elif element_type
== "VNF":
1539 num_units
= db_vnfr
.get("config-units") or 1
1540 elif element_type
== "VDU":
1541 for v
in db_vnfr
["vdur"]:
1542 if vdu_id
== v
["vdu-id-ref"]:
1543 num_units
= v
.get("config-units") or 1
1546 await self
.vca_map
[vca_type
].install_configuration_sw(
1548 artifact_path
=artifact_path
,
1551 num_units
=num_units
,
1555 # write in db flag of configuration_sw already installed
1556 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1558 # add relations for this VCA (wait for other peers related with this VCA)
1559 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
,
1560 vca_index
=vca_index
, vca_type
=vca_type
)
1562 # if SSH access is required, then get execution environment SSH public
1563 # if native charm we have waited already to VM be UP
1564 if vca_type
in ("k8s_proxy_charm", "lxc_proxy_charm", "helm"):
1567 # self.logger.debug("get ssh key block")
1568 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1569 # self.logger.debug("ssh key needed")
1570 # Needed to inject a ssh key
1571 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1572 step
= "Install configuration Software, getting public ssh key"
1573 pub_key
= await self
.vca_map
[vca_type
].get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1575 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1577 # self.logger.debug("no need to get ssh key")
1578 step
= "Waiting to VM being up and getting IP address"
1579 self
.logger
.debug(logging_text
+ step
)
1581 # n2vc_redesign STEP 5.1
1582 # wait for RO (ip-address) Insert pub_key into VM
1584 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1585 user
=user
, pub_key
=pub_key
)
1587 rw_mgmt_ip
= None # This is for a NS configuration
1589 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1591 # store rw_mgmt_ip in deploy params for later replacement
1592 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1594 # n2vc_redesign STEP 6 Execute initial config primitive
1595 step
= 'execute initial config primitive'
1597 # wait for dependent primitives execution (NS -> VNF -> VDU)
1598 if initial_config_primitive_list
:
1599 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1601 # stage, in function of element type: vdu, kdu, vnf or ns
1602 my_vca
= vca_deployed_list
[vca_index
]
1603 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1605 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1606 elif my_vca
.get("member-vnf-index"):
1608 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1611 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1613 self
._write
_configuration
_status
(
1615 vca_index
=vca_index
,
1616 status
='EXECUTING PRIMITIVE'
1619 self
._write
_op
_status
(
1624 check_if_terminated_needed
= True
1625 for initial_config_primitive
in initial_config_primitive_list
:
1626 # adding information on the vca_deployed if it is a NS execution environment
1627 if not vca_deployed
["member-vnf-index"]:
1628 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1629 # TODO check if already done
1630 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1632 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1633 self
.logger
.debug(logging_text
+ step
)
1634 await self
.vca_map
[vca_type
].exec_primitive(
1636 primitive_name
=initial_config_primitive
["name"],
1637 params_dict
=primitive_params_
,
1640 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1641 if check_if_terminated_needed
:
1642 if config_descriptor
.get('terminate-config-primitive'):
1643 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1644 check_if_terminated_needed
= False
1646 # TODO register in database that primitive is done
1648 # STEP 7 Configure metrics
1649 if vca_type
== "helm":
1650 prometheus_jobs
= await self
.add_prometheus_metrics(
1652 artifact_path
=artifact_path
,
1653 ee_config_descriptor
=ee_config_descriptor
,
1656 target_ip
=rw_mgmt_ip
,
1659 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "prometheus_jobs": prometheus_jobs
})
1661 step
= "instantiated at VCA"
1662 self
.logger
.debug(logging_text
+ step
)
1664 self
._write
_configuration
_status
(
1666 vca_index
=vca_index
,
1670 except Exception as e
: # TODO not use Exception but N2VC exception
1671 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1672 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1673 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1674 self
._write
_configuration
_status
(
1676 vca_index
=vca_index
,
1679 raise LcmException("{} {}".format(step
, e
)) from e
1681 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1682 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1684 Update db_nsr fields.
1687 :param current_operation:
1688 :param current_operation_id:
1689 :param error_description:
1690 :param error_detail:
1691 :param other_update: Other required changes at database if provided, will be cleared
1695 db_dict
= other_update
or {}
1696 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1697 db_dict
["_admin.current-operation"] = current_operation_id
1698 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1699 db_dict
["currentOperation"] = current_operation
1700 db_dict
["currentOperationID"] = current_operation_id
1701 db_dict
["errorDescription"] = error_description
1702 db_dict
["errorDetail"] = error_detail
1705 db_dict
["nsState"] = ns_state
1706 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1707 except DbException
as e
:
1708 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1710 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1711 operation_state
: str = None, other_update
: dict = None):
1713 db_dict
= other_update
or {}
1714 db_dict
['queuePosition'] = queuePosition
1715 if isinstance(stage
, list):
1716 db_dict
['stage'] = stage
[0]
1717 db_dict
['detailed-status'] = " ".join(stage
)
1718 elif stage
is not None:
1719 db_dict
['stage'] = str(stage
)
1721 if error_message
is not None:
1722 db_dict
['errorMessage'] = error_message
1723 if operation_state
is not None:
1724 db_dict
['operationState'] = operation_state
1725 db_dict
["statusEnteredTime"] = time()
1726 self
.update_db_2("nslcmops", op_id
, db_dict
)
1727 except DbException
as e
:
1728 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1730 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1732 nsr_id
= db_nsr
["_id"]
1733 # configurationStatus
1734 config_status
= db_nsr
.get('configurationStatus')
1736 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1737 enumerate(config_status
) if v
}
1739 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1741 except DbException
as e
:
1742 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1744 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1745 element_under_configuration
: str = None, element_type
: str = None,
1746 other_update
: dict = None):
1748 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1749 # .format(vca_index, status))
1752 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1753 db_dict
= other_update
or {}
1755 db_dict
[db_path
+ 'status'] = status
1756 if element_under_configuration
:
1757 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1759 db_dict
[db_path
+ 'elementType'] = element_type
1760 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1761 except DbException
as e
:
1762 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1763 .format(status
, nsr_id
, vca_index
, e
))
1765 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1767 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1768 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1769 Database is used because the result can be obtained from a different LCM worker in case of HA.
1770 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1771 :param db_nslcmop: database content of nslcmop
1772 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1773 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
1774 computed 'vim-account-id'
1777 nslcmop_id
= db_nslcmop
['_id']
1778 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1779 if placement_engine
== "PLA":
1780 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1781 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1782 db_poll_interval
= 5
1783 wait
= db_poll_interval
* 10
1785 while not pla_result
and wait
>= 0:
1786 await asyncio
.sleep(db_poll_interval
)
1787 wait
-= db_poll_interval
1788 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1789 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1792 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1794 for pla_vnf
in pla_result
['vnf']:
1795 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1796 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1799 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1801 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1804 def update_nsrs_with_pla_result(self
, params
):
1806 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1807 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1808 except Exception as e
:
1809 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1811 async def instantiate(self
, nsr_id
, nslcmop_id
):
1814 :param nsr_id: ns instance to deploy
1815 :param nslcmop_id: operation to run
1819 # Try to lock HA task here
1820 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1821 if not task_is_locked_by_me
:
1822 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1825 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1826 self
.logger
.debug(logging_text
+ "Enter")
1828 # get all needed from database
1830 # database nsrs record
1833 # database nslcmops record
1836 # update operation on nsrs
1838 # update operation on nslcmops
1839 db_nslcmop_update
= {}
1841 nslcmop_operation_state
= None
1842 db_vnfrs
= {} # vnf's info indexed by member-index
1844 tasks_dict_info
= {} # from task to info text
1847 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1848 # ^ stage, step, VIM progress
1850 # wait for any previous tasks in process
1851 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1853 stage
[1] = "Sync filesystem from database."
1854 self
.fs
.sync() # TODO, make use of partial sync, only for the needed packages
1856 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1857 stage
[1] = "Reading from database."
1858 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1859 db_nsr_update
["detailed-status"] = "creating"
1860 db_nsr_update
["operational-status"] = "init"
1861 self
._write
_ns
_status
(
1863 ns_state
="BUILDING",
1864 current_operation
="INSTANTIATING",
1865 current_operation_id
=nslcmop_id
,
1866 other_update
=db_nsr_update
1868 self
._write
_op
_status
(
1874 # read from db: operation
1875 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
1876 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1877 ns_params
= db_nslcmop
.get("operationParams")
1878 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1879 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1881 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1884 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
1885 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1886 stage
[1] = "Getting nsd={} from db.".format(db_nsr
["nsd-id"])
1887 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
1889 # nsr_name = db_nsr["name"] # TODO short-name??
1891 # read from db: vnf's of this ns
1892 stage
[1] = "Getting vnfrs from db."
1893 self
.logger
.debug(logging_text
+ stage
[1])
1894 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1896 # read from db: vnfd's for every vnf
1897 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1898 db_vnfds
= {} # every vnfd data indexed by vnf id
1899 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1901 # for each vnf in ns, read vnfd
1902 for vnfr
in db_vnfrs_list
:
1903 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1904 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1905 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1907 # if we haven't this vnfd, read it from db
1908 if vnfd_id
not in db_vnfds
:
1910 stage
[1] = "Getting vnfd={} id='{}' from db.".format(vnfd_id
, vnfd_ref
)
1911 self
.logger
.debug(logging_text
+ stage
[1])
1912 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1915 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1916 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1917 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1919 # Get or generates the _admin.deployed.VCA list
1920 vca_deployed_list
= None
1921 if db_nsr
["_admin"].get("deployed"):
1922 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1923 if vca_deployed_list
is None:
1924 vca_deployed_list
= []
1925 configuration_status_list
= []
1926 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1927 db_nsr_update
["configurationStatus"] = configuration_status_list
1928 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1929 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1930 elif isinstance(vca_deployed_list
, dict):
1931 # maintain backward compatibility. Change a dict to list at database
1932 vca_deployed_list
= list(vca_deployed_list
.values())
1933 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1934 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1936 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1937 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1938 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1940 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1941 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1942 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1943 self
.db
.set_list("vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "INSTANTIATED"})
1945 # n2vc_redesign STEP 2 Deploy Network Scenario
1946 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1947 self
._write
_op
_status
(
1952 stage
[1] = "Deploying KDUs."
1953 # self.logger.debug(logging_text + "Before deploy_kdus")
1954 # Call to deploy_kdus in case exists the "vdu:kdu" param
1955 await self
.deploy_kdus(
1956 logging_text
=logging_text
,
1958 nslcmop_id
=nslcmop_id
,
1961 task_instantiation_info
=tasks_dict_info
,
1964 stage
[1] = "Getting VCA public key."
1965 # n2vc_redesign STEP 1 Get VCA public ssh-key
1966 # feature 1429. Add n2vc public key to needed VMs
1967 n2vc_key
= self
.n2vc
.get_public_key()
1968 n2vc_key_list
= [n2vc_key
]
1969 if self
.vca_config
.get("public_key"):
1970 n2vc_key_list
.append(self
.vca_config
["public_key"])
1972 stage
[1] = "Deploying NS at VIM."
1973 task_ro
= asyncio
.ensure_future(
1974 self
.instantiate_RO(
1975 logging_text
=logging_text
,
1979 db_nslcmop
=db_nslcmop
,
1981 db_vnfds_ref
=db_vnfds_ref
,
1982 n2vc_key_list
=n2vc_key_list
,
1986 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
1987 tasks_dict_info
[task_ro
] = "Deploying at VIM"
1989 # n2vc_redesign STEP 3 to 6 Deploy N2VC
1990 stage
[1] = "Deploying Execution Environments."
1991 self
.logger
.debug(logging_text
+ stage
[1])
1993 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
1994 # get_iterable() returns a value from a dict or empty tuple if key does not exist
1995 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
1996 vnfd_id
= c_vnf
["vnfd-id-ref"]
1997 vnfd
= db_vnfds_ref
[vnfd_id
]
1998 member_vnf_index
= str(c_vnf
["member-vnf-index"])
1999 db_vnfr
= db_vnfrs
[member_vnf_index
]
2000 base_folder
= vnfd
["_admin"]["storage"]
2006 # Get additional parameters
2008 if db_vnfr
.get("additionalParamsForVnf"):
2009 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
2011 descriptor_config
= vnfd
.get("vnf-configuration")
2012 if descriptor_config
:
2014 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
2017 nslcmop_id
=nslcmop_id
,
2023 member_vnf_index
=member_vnf_index
,
2024 vdu_index
=vdu_index
,
2026 deploy_params
=deploy_params
,
2027 descriptor_config
=descriptor_config
,
2028 base_folder
=base_folder
,
2029 task_instantiation_info
=tasks_dict_info
,
2033 # Deploy charms for each VDU that supports one.
2034 for vdud
in get_iterable(vnfd
, 'vdu'):
2036 descriptor_config
= vdud
.get('vdu-configuration')
2037 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
2038 if vdur
.get("additionalParams"):
2039 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
2041 deploy_params_vdu
= deploy_params
2042 if descriptor_config
:
2043 # look for vdu index in the db_vnfr["vdu"] section
2044 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
2045 # if vdur["vdu-id-ref"] == vdu_id:
2048 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
2049 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
2050 # vdu_name = vdur.get("name")
2053 for vdu_index
in range(int(vdud
.get("count", 1))):
2054 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
2056 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
2057 member_vnf_index
, vdu_id
, vdu_index
),
2060 nslcmop_id
=nslcmop_id
,
2066 member_vnf_index
=member_vnf_index
,
2067 vdu_index
=vdu_index
,
2069 deploy_params
=deploy_params_vdu
,
2070 descriptor_config
=descriptor_config
,
2071 base_folder
=base_folder
,
2072 task_instantiation_info
=tasks_dict_info
,
2075 for kdud
in get_iterable(vnfd
, 'kdu'):
2076 kdu_name
= kdud
["name"]
2077 descriptor_config
= kdud
.get('kdu-configuration')
2078 if descriptor_config
:
2082 # look for vdu index in the db_vnfr["vdu"] section
2083 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
2084 # if vdur["vdu-id-ref"] == vdu_id:
2087 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
2088 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
2089 # vdu_name = vdur.get("name")
2093 logging_text
=logging_text
,
2096 nslcmop_id
=nslcmop_id
,
2102 member_vnf_index
=member_vnf_index
,
2103 vdu_index
=vdu_index
,
2105 deploy_params
=deploy_params
,
2106 descriptor_config
=descriptor_config
,
2107 base_folder
=base_folder
,
2108 task_instantiation_info
=tasks_dict_info
,
2112 # Check if this NS has a charm configuration
2113 descriptor_config
= nsd
.get("ns-configuration")
2114 if descriptor_config
and descriptor_config
.get("juju"):
2117 member_vnf_index
= None
2123 # Get additional parameters
2125 if db_nsr
.get("additionalParamsForNs"):
2126 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
2127 base_folder
= nsd
["_admin"]["storage"]
2129 logging_text
=logging_text
,
2132 nslcmop_id
=nslcmop_id
,
2138 member_vnf_index
=member_vnf_index
,
2139 vdu_index
=vdu_index
,
2141 deploy_params
=deploy_params
,
2142 descriptor_config
=descriptor_config
,
2143 base_folder
=base_folder
,
2144 task_instantiation_info
=tasks_dict_info
,
2148 # rest of staff will be done at finally
2150 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2151 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
2153 except asyncio
.CancelledError
:
2154 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2155 exc
= "Operation was cancelled"
2156 except Exception as e
:
2157 exc
= traceback
.format_exc()
2158 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2161 error_list
.append(str(exc
))
2163 # wait for pending tasks
2165 stage
[1] = "Waiting for instantiate pending tasks."
2166 self
.logger
.debug(logging_text
+ stage
[1])
2167 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
2168 stage
, nslcmop_id
, nsr_id
=nsr_id
)
2169 stage
[1] = stage
[2] = ""
2170 except asyncio
.CancelledError
:
2171 error_list
.append("Cancelled")
2172 # TODO cancel all tasks
2173 except Exception as exc
:
2174 error_list
.append(str(exc
))
2176 # update operation-status
2177 db_nsr_update
["operational-status"] = "running"
2178 # let's begin with VCA 'configured' status (later we can change it)
2179 db_nsr_update
["config-status"] = "configured"
2180 for task
, task_name
in tasks_dict_info
.items():
2181 if not task
.done() or task
.cancelled() or task
.exception():
2182 if task_name
.startswith(self
.task_name_deploy_vca
):
2183 # A N2VC task is pending
2184 db_nsr_update
["config-status"] = "failed"
2186 # RO or KDU task is pending
2187 db_nsr_update
["operational-status"] = "failed"
2189 # update status at database
2191 error_detail
= ". ".join(error_list
)
2192 self
.logger
.error(logging_text
+ error_detail
)
2193 error_description_nslcmop
= '{} Detail: {}'.format(stage
[0], error_detail
)
2194 error_description_nsr
= 'Operation: INSTANTIATING.{}, {}'.format(nslcmop_id
, stage
[0])
2196 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2197 db_nslcmop_update
["detailed-status"] = error_detail
2198 nslcmop_operation_state
= "FAILED"
2202 error_description_nsr
= error_description_nslcmop
= None
2204 db_nsr_update
["detailed-status"] = "Done"
2205 db_nslcmop_update
["detailed-status"] = "Done"
2206 nslcmop_operation_state
= "COMPLETED"
2209 self
._write
_ns
_status
(
2212 current_operation
="IDLE",
2213 current_operation_id
=None,
2214 error_description
=error_description_nsr
,
2215 error_detail
=error_detail
,
2216 other_update
=db_nsr_update
2218 self
._write
_op
_status
(
2221 error_message
=error_description_nslcmop
,
2222 operation_state
=nslcmop_operation_state
,
2223 other_update
=db_nslcmop_update
,
2226 if nslcmop_operation_state
:
2228 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2229 "operationState": nslcmop_operation_state
},
2231 except Exception as e
:
2232 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2234 self
.logger
.debug(logging_text
+ "Exit")
2235 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2237 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int,
2238 timeout
: int = 3600, vca_type
: str = None) -> bool:
2241 # 1. find all relations for this VCA
2242 # 2. wait for other peers related
2246 vca_type
= vca_type
or "lxc_proxy_charm"
2248 # STEP 1: find all relations for this VCA
2251 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2252 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2255 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
2257 # read all ns-configuration relations
2258 ns_relations
= list()
2259 db_ns_relations
= deep_get(nsd
, ('ns-configuration', 'relation'))
2261 for r
in db_ns_relations
:
2262 # check if this VCA is in the relation
2263 if my_vca
.get('member-vnf-index') in\
2264 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2265 ns_relations
.append(r
)
2267 # read all vnf-configuration relations
2268 vnf_relations
= list()
2269 db_vnfd_list
= db_nsr
.get('vnfd-id')
2271 for vnfd
in db_vnfd_list
:
2272 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2273 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
2274 if db_vnf_relations
:
2275 for r
in db_vnf_relations
:
2276 # check if this VCA is in the relation
2277 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2278 vnf_relations
.append(r
)
2280 # if no relations, terminate
2281 if not ns_relations
and not vnf_relations
:
2282 self
.logger
.debug(logging_text
+ ' No relations')
2285 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
2292 if now
- start
>= timeout
:
2293 self
.logger
.error(logging_text
+ ' : timeout adding relations')
2296 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2297 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2299 # for each defined NS relation, find the VCA's related
2300 for r
in ns_relations
:
2301 from_vca_ee_id
= None
2303 from_vca_endpoint
= None
2304 to_vca_endpoint
= None
2305 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2306 for vca
in vca_list
:
2307 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
2308 and vca
.get('config_sw_installed'):
2309 from_vca_ee_id
= vca
.get('ee_id')
2310 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2311 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2312 and vca
.get('config_sw_installed'):
2313 to_vca_ee_id
= vca
.get('ee_id')
2314 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2315 if from_vca_ee_id
and to_vca_ee_id
:
2317 await self
.vca_map
[vca_type
].add_relation(
2318 ee_id_1
=from_vca_ee_id
,
2319 ee_id_2
=to_vca_ee_id
,
2320 endpoint_1
=from_vca_endpoint
,
2321 endpoint_2
=to_vca_endpoint
)
2322 # remove entry from relations list
2323 ns_relations
.remove(r
)
2325 # check failed peers
2327 vca_status_list
= db_nsr
.get('configurationStatus')
2329 for i
in range(len(vca_list
)):
2331 vca_status
= vca_status_list
[i
]
2332 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2333 if vca_status
.get('status') == 'BROKEN':
2334 # peer broken: remove relation from list
2335 ns_relations
.remove(r
)
2336 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2337 if vca_status
.get('status') == 'BROKEN':
2338 # peer broken: remove relation from list
2339 ns_relations
.remove(r
)
2344 # for each defined VNF relation, find the VCA's related
2345 for r
in vnf_relations
:
2346 from_vca_ee_id
= None
2348 from_vca_endpoint
= None
2349 to_vca_endpoint
= None
2350 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2351 for vca
in vca_list
:
2352 if vca
.get('vdu_id') == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2353 from_vca_ee_id
= vca
.get('ee_id')
2354 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2355 if vca
.get('vdu_id') == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2356 to_vca_ee_id
= vca
.get('ee_id')
2357 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2358 if from_vca_ee_id
and to_vca_ee_id
:
2360 await self
.vca_map
[vca_type
].add_relation(
2361 ee_id_1
=from_vca_ee_id
,
2362 ee_id_2
=to_vca_ee_id
,
2363 endpoint_1
=from_vca_endpoint
,
2364 endpoint_2
=to_vca_endpoint
)
2365 # remove entry from relations list
2366 vnf_relations
.remove(r
)
2368 # check failed peers
2370 vca_status_list
= db_nsr
.get('configurationStatus')
2372 for i
in range(len(vca_list
)):
2374 vca_status
= vca_status_list
[i
]
2375 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2376 if vca_status
.get('status') == 'BROKEN':
2377 # peer broken: remove relation from list
2378 ns_relations
.remove(r
)
2379 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2380 if vca_status
.get('status') == 'BROKEN':
2381 # peer broken: remove relation from list
2382 ns_relations
.remove(r
)
2388 await asyncio
.sleep(5.0)
2390 if not ns_relations
and not vnf_relations
:
2391 self
.logger
.debug('Relations added')
2396 except Exception as e
:
2397 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2400 async def _install_kdu(self
, nsr_id
: str, nsr_db_path
: str, vnfr_data
: dict, kdu_index
: int, kdur
: dict, kdud
: dict,
2401 vnfd
: dict, k8s_instance_info
: dict, k8params
: dict = None, timeout
: int = 600):
2404 k8sclustertype
= k8s_instance_info
["k8scluster-type"]
2406 db_dict_install
= {"collection": "nsrs",
2407 "filter": {"_id": nsr_id
},
2408 "path": nsr_db_path
}
2410 kdu_instance
= await self
.k8scluster_map
[k8sclustertype
].install(
2411 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2412 kdu_model
=k8s_instance_info
["kdu-model"],
2415 db_dict
=db_dict_install
,
2417 kdu_name
=k8s_instance_info
["kdu-name"],
2418 namespace
=k8s_instance_info
["namespace"])
2419 self
.update_db_2("nsrs", nsr_id
, {nsr_db_path
+ ".kdu-instance": kdu_instance
})
2421 # Obtain services to obtain management service ip
2422 services
= await self
.k8scluster_map
[k8sclustertype
].get_services(
2423 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2424 kdu_instance
=kdu_instance
,
2425 namespace
=k8s_instance_info
["namespace"])
2427 # Obtain management service info (if exists)
2429 vnfr_update_dict
= {"kdur.{}.services".format(kdu_index
): services
}
2430 mgmt_services
= [service
for service
in kdud
.get("service", []) if service
.get("mgmt-service")]
2431 for mgmt_service
in mgmt_services
:
2432 for service
in services
:
2433 if service
["name"].startswith(mgmt_service
["name"]):
2434 # Mgmt service found, Obtain service ip
2435 ip
= service
.get("external_ip", service
.get("cluster_ip"))
2436 if isinstance(ip
, list) and len(ip
) == 1:
2439 vnfr_update_dict
["kdur.{}.ip-address".format(kdu_index
)] = ip
2441 # Check if must update also mgmt ip at the vnf
2442 service_external_cp
= mgmt_service
.get("external-connection-point-ref")
2443 if service_external_cp
:
2444 if deep_get(vnfd
, ("mgmt-interface", "cp")) == service_external_cp
:
2445 vnfr_update_dict
["ip-address"] = ip
2449 self
.logger
.warn("Mgmt service name: {} not found".format(mgmt_service
["name"]))
2451 self
.update_db_2("vnfrs", vnfr_data
.get("_id"), vnfr_update_dict
)
2453 kdu_config
= kdud
.get("kdu-configuration")
2454 if kdu_config
and kdu_config
.get("initial-config-primitive") and kdu_config
.get("juju") is None:
2455 initial_config_primitive_list
= kdu_config
.get("initial-config-primitive")
2456 initial_config_primitive_list
.sort(key
=lambda val
: int(val
["seq"]))
2458 for initial_config_primitive
in initial_config_primitive_list
:
2459 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, {})
2461 await asyncio
.wait_for(
2462 self
.k8scluster_map
[k8sclustertype
].exec_primitive(
2463 cluster_uuid
=k8s_instance_info
["k8scluster-uuid"],
2464 kdu_instance
=kdu_instance
,
2465 primitive_name
=initial_config_primitive
["name"],
2466 params
=primitive_params_
, db_dict
={}),
2469 except Exception as e
:
2470 # Prepare update db with error and raise exception
2472 self
.update_db_2("nsrs", nsr_id
, {nsr_db_path
+ ".detailed-status": str(e
)})
2474 # ignore to keep original exception
2476 # reraise original error
2481 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2482 # Launch kdus if present in the descriptor
2484 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2486 async def _get_cluster_id(cluster_id
, cluster_type
):
2487 nonlocal k8scluster_id_2_uuic
2488 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2489 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2491 # check if K8scluster is creating and wait look if previous tasks in process
2492 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("k8scluster", cluster_id
)
2494 text
= "Waiting for related tasks '{}' on k8scluster {} to be completed".format(task_name
, cluster_id
)
2495 self
.logger
.debug(logging_text
+ text
)
2496 await asyncio
.wait(task_dependency
, timeout
=3600)
2498 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2499 if not db_k8scluster
:
2500 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2502 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2504 raise LcmException("K8s cluster '{}' has not been initialized for '{}'".format(cluster_id
,
2506 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2509 logging_text
+= "Deploy kdus: "
2512 db_nsr_update
= {"_admin.deployed.K8s": []}
2513 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2516 updated_cluster_list
= []
2518 for vnfr_data
in db_vnfrs
.values():
2519 for kdu_index
, kdur
in enumerate(get_iterable(vnfr_data
, "kdur")):
2520 # Step 0: Prepare and set parameters
2521 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2522 vnfd_id
= vnfr_data
.get('vnfd-id')
2523 kdud
= next(kdud
for kdud
in db_vnfds
[vnfd_id
]["kdu"] if kdud
["name"] == kdur
["kdu-name"])
2524 namespace
= kdur
.get("k8s-namespace")
2525 if kdur
.get("helm-chart"):
2526 kdumodel
= kdur
["helm-chart"]
2527 k8sclustertype
= "helm-chart"
2528 elif kdur
.get("juju-bundle"):
2529 kdumodel
= kdur
["juju-bundle"]
2530 k8sclustertype
= "juju-bundle"
2532 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2533 "juju-bundle. Maybe an old NBI version is running".
2534 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2535 # check if kdumodel is a file and exists
2537 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2538 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2539 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2540 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["pkg-dir"], k8sclustertype
,
2542 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2543 kdumodel
= self
.fs
.path
+ filename
2544 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2546 except Exception: # it is not a file
2549 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2550 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2551 cluster_uuid
= await _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2554 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2555 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2556 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2557 if del_repo_list
or added_repo_dict
:
2558 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2559 updated
= {'_admin.helm_charts_added.' +
2560 item
: name
for item
, name
in added_repo_dict
.items()}
2561 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2562 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2564 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2565 updated_cluster_list
.append(cluster_uuid
)
2568 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2569 kdur
["kdu-name"], k8s_cluster_id
)
2570 k8s_instance_info
= {"kdu-instance": None,
2571 "k8scluster-uuid": cluster_uuid
,
2572 "k8scluster-type": k8sclustertype
,
2573 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2574 "kdu-name": kdur
["kdu-name"],
2575 "kdu-model": kdumodel
,
2576 "namespace": namespace
}
2577 db_path
= "_admin.deployed.K8s.{}".format(index
)
2578 db_nsr_update
[db_path
] = k8s_instance_info
2579 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2581 task
= asyncio
.ensure_future(
2582 self
._install
_kdu
(nsr_id
, db_path
, vnfr_data
, kdu_index
, kdur
, kdud
, db_vnfds
[vnfd_id
],
2583 k8s_instance_info
, k8params
=desc_params
, timeout
=600))
2584 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2585 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2589 except (LcmException
, asyncio
.CancelledError
):
2591 except Exception as e
:
2592 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2593 if isinstance(e
, (N2VCException
, DbException
)):
2594 self
.logger
.error(logging_text
+ msg
)
2596 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2597 raise LcmException(msg
)
2600 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2602 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2603 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2604 base_folder
, task_instantiation_info
, stage
):
2605 # launch instantiate_N2VC in a asyncio task and register task object
2606 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2607 # if not found, create one entry and update database
2608 # fill db_nsr._admin.deployed.VCA.<index>
2610 self
.logger
.debug(logging_text
+ "_deploy_n2vc vnfd_id={}, vdu_id={}".format(vnfd_id
, vdu_id
))
2611 if descriptor_config
.get("juju"): # There is one execution envioronment of type juju
2612 ee_list
= [descriptor_config
]
2613 elif descriptor_config
.get("execution-environment-list"):
2614 ee_list
= descriptor_config
.get("execution-environment-list")
2615 else: # other types as script are not supported
2618 for ee_item
in ee_list
:
2619 self
.logger
.debug(logging_text
+ "_deploy_n2vc ee_item juju={}, helm={}".format(ee_item
.get('juju'),
2620 ee_item
.get("helm-chart")))
2621 ee_descriptor_id
= ee_item
.get("id")
2622 if ee_item
.get("juju"):
2623 vca_name
= ee_item
['juju'].get('charm')
2624 vca_type
= "lxc_proxy_charm" if ee_item
['juju'].get('charm') is not None else "native_charm"
2625 if ee_item
['juju'].get('cloud') == "k8s":
2626 vca_type
= "k8s_proxy_charm"
2627 elif ee_item
['juju'].get('proxy') is False:
2628 vca_type
= "native_charm"
2629 elif ee_item
.get("helm-chart"):
2630 vca_name
= ee_item
['helm-chart']
2633 self
.logger
.debug(logging_text
+ "skipping non juju neither charm configuration")
2637 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2638 if not vca_deployed
:
2640 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2641 vca_deployed
.get("vdu_id") == vdu_id
and \
2642 vca_deployed
.get("kdu_name") == kdu_name
and \
2643 vca_deployed
.get("vdu_count_index", 0) == vdu_index
and \
2644 vca_deployed
.get("ee_descriptor_id") == ee_descriptor_id
:
2647 # not found, create one.
2648 target
= "ns" if not member_vnf_index
else "vnf/{}".format(member_vnf_index
)
2650 target
+= "/vdu/{}/{}".format(vdu_id
, vdu_index
or 0)
2652 target
+= "/kdu/{}".format(kdu_name
)
2654 "target_element": target
,
2655 # ^ target_element will replace member-vnf-index, kdu_name, vdu_id ... in a single string
2656 "member-vnf-index": member_vnf_index
,
2658 "kdu_name": kdu_name
,
2659 "vdu_count_index": vdu_index
,
2660 "operational-status": "init", # TODO revise
2661 "detailed-status": "", # TODO revise
2662 "step": "initial-deploy", # TODO revise
2664 "vdu_name": vdu_name
,
2666 "ee_descriptor_id": ee_descriptor_id
2670 # create VCA and configurationStatus in db
2672 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2673 "configurationStatus.{}".format(vca_index
): dict()
2675 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2677 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2680 task_n2vc
= asyncio
.ensure_future(
2681 self
.instantiate_N2VC(
2682 logging_text
=logging_text
,
2683 vca_index
=vca_index
,
2689 vdu_index
=vdu_index
,
2690 deploy_params
=deploy_params
,
2691 config_descriptor
=descriptor_config
,
2692 base_folder
=base_folder
,
2693 nslcmop_id
=nslcmop_id
,
2697 ee_config_descriptor
=ee_item
2700 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2701 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2702 member_vnf_index
or "", vdu_id
or "")
2705 def _get_terminate_config_primitive(primitive_list
, vca_deployed
):
2706 """ Get a sorted terminate config primitive list. In case ee_descriptor_id is present at vca_deployed,
2707 it get only those primitives for this execution envirom"""
2709 primitive_list
= primitive_list
or []
2710 # filter primitives by ee_descriptor_id
2711 ee_descriptor_id
= vca_deployed
.get("ee_descriptor_id")
2712 primitive_list
= [p
for p
in primitive_list
if p
.get("execution-environment-ref") == ee_descriptor_id
]
2715 primitive_list
.sort(key
=lambda val
: int(val
['seq']))
2717 return primitive_list
2720 def _create_nslcmop(nsr_id
, operation
, params
):
2722 Creates a ns-lcm-opp content to be stored at database.
2723 :param nsr_id: internal id of the instance
2724 :param operation: instantiate, terminate, scale, action, ...
2725 :param params: user parameters for the operation
2726 :return: dictionary following SOL005 format
2728 # Raise exception if invalid arguments
2729 if not (nsr_id
and operation
and params
):
2731 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2737 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2738 "operationState": "PROCESSING",
2739 "statusEnteredTime": now
,
2740 "nsInstanceId": nsr_id
,
2741 "lcmOperationType": operation
,
2743 "isAutomaticInvocation": False,
2744 "operationParams": params
,
2745 "isCancelPending": False,
2747 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2748 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2753 def _format_additional_params(self
, params
):
2754 params
= params
or {}
2755 for key
, value
in params
.items():
2756 if str(value
).startswith("!!yaml "):
2757 params
[key
] = yaml
.safe_load(value
[7:])
2760 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2761 primitive
= seq
.get('name')
2762 primitive_params
= {}
2764 "member_vnf_index": vnf_index
,
2765 "primitive": primitive
,
2766 "primitive_params": primitive_params
,
2769 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2773 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2774 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2775 if op
.get('operationState') == 'COMPLETED':
2776 # b. Skip sub-operation
2777 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2778 return self
.SUBOPERATION_STATUS_SKIP
2780 # c. retry executing sub-operation
2781 # The sub-operation exists, and operationState != 'COMPLETED'
2782 # Update operationState = 'PROCESSING' to indicate a retry.
2783 operationState
= 'PROCESSING'
2784 detailed_status
= 'In progress'
2785 self
._update
_suboperation
_status
(
2786 db_nslcmop
, op_index
, operationState
, detailed_status
)
2787 # Return the sub-operation index
2788 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2789 # with arguments extracted from the sub-operation
2792 # Find a sub-operation where all keys in a matching dictionary must match
2793 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2794 def _find_suboperation(self
, db_nslcmop
, match
):
2795 if db_nslcmop
and match
:
2796 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2797 for i
, op
in enumerate(op_list
):
2798 if all(op
.get(k
) == match
[k
] for k
in match
):
2800 return self
.SUBOPERATION_STATUS_NOT_FOUND
2802 # Update status for a sub-operation given its index
2803 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2804 # Update DB for HA tasks
2805 q_filter
= {'_id': db_nslcmop
['_id']}
2806 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2807 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2808 self
.db
.set_one("nslcmops",
2810 update_dict
=update_dict
,
2811 fail_on_empty
=False)
2813 # Add sub-operation, return the index of the added sub-operation
2814 # Optionally, set operationState, detailed-status, and operationType
2815 # Status and type are currently set for 'scale' sub-operations:
2816 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2817 # 'detailed-status' : status message
2818 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2819 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2820 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2821 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2822 RO_nsr_id
=None, RO_scaling_info
=None):
2824 return self
.SUBOPERATION_STATUS_NOT_FOUND
2825 # Get the "_admin.operations" list, if it exists
2826 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2827 op_list
= db_nslcmop_admin
.get('operations')
2828 # Create or append to the "_admin.operations" list
2829 new_op
= {'member_vnf_index': vnf_index
,
2831 'vdu_count_index': vdu_count_index
,
2832 'primitive': primitive
,
2833 'primitive_params': mapped_primitive_params
}
2835 new_op
['operationState'] = operationState
2837 new_op
['detailed-status'] = detailed_status
2839 new_op
['lcmOperationType'] = operationType
2841 new_op
['RO_nsr_id'] = RO_nsr_id
2843 new_op
['RO_scaling_info'] = RO_scaling_info
2845 # No existing operations, create key 'operations' with current operation as first list element
2846 db_nslcmop_admin
.update({'operations': [new_op
]})
2847 op_list
= db_nslcmop_admin
.get('operations')
2849 # Existing operations, append operation to list
2850 op_list
.append(new_op
)
2852 db_nslcmop_update
= {'_admin.operations': op_list
}
2853 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2854 op_index
= len(op_list
) - 1
2857 # Helper methods for scale() sub-operations
2859 # pre-scale/post-scale:
2860 # Check for 3 different cases:
2861 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2862 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2863 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2864 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2865 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2866 # Find this sub-operation
2867 if RO_nsr_id
and RO_scaling_info
:
2868 operationType
= 'SCALE-RO'
2870 'member_vnf_index': vnf_index
,
2871 'RO_nsr_id': RO_nsr_id
,
2872 'RO_scaling_info': RO_scaling_info
,
2876 'member_vnf_index': vnf_index
,
2877 'primitive': vnf_config_primitive
,
2878 'primitive_params': primitive_params
,
2879 'lcmOperationType': operationType
2881 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2882 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2883 # a. New sub-operation
2884 # The sub-operation does not exist, add it.
2885 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2886 # The following parameters are set to None for all kind of scaling:
2888 vdu_count_index
= None
2890 if RO_nsr_id
and RO_scaling_info
:
2891 vnf_config_primitive
= None
2892 primitive_params
= None
2895 RO_scaling_info
= None
2896 # Initial status for sub-operation
2897 operationState
= 'PROCESSING'
2898 detailed_status
= 'In progress'
2899 # Add sub-operation for pre/post-scaling (zero or more operations)
2900 self
._add
_suboperation
(db_nslcmop
,
2905 vnf_config_primitive
,
2912 return self
.SUBOPERATION_STATUS_NEW
2914 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2915 # or op_index (operationState != 'COMPLETED')
2916 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2918 # Function to return execution_environment id
2920 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2921 # TODO vdu_index_count
2922 for vca
in vca_deployed_list
:
2923 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2926 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
,
2927 vca_index
, destroy_ee
=True, exec_primitives
=True):
2929 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2930 :param logging_text:
2932 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2933 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2934 :param vca_index: index in the database _admin.deployed.VCA
2935 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2936 :param exec_primitives: False to do not execute terminate primitives, because the config is not completed or has
2937 not executed properly
2938 :return: None or exception
2942 logging_text
+ " vca_index: {}, vca_deployed: {}, config_descriptor: {}, destroy_ee: {}".format(
2943 vca_index
, vca_deployed
, config_descriptor
, destroy_ee
2947 vca_type
= vca_deployed
.get("type", "lxc_proxy_charm")
2949 # execute terminate_primitives
2951 terminate_primitives
= self
._get
_terminate
_config
_primitive
(
2952 config_descriptor
.get("terminate-config-primitive"), vca_deployed
)
2953 vdu_id
= vca_deployed
.get("vdu_id")
2954 vdu_count_index
= vca_deployed
.get("vdu_count_index")
2955 vdu_name
= vca_deployed
.get("vdu_name")
2956 vnf_index
= vca_deployed
.get("member-vnf-index")
2957 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
2958 for seq
in terminate_primitives
:
2959 # For each sequence in list, get primitive and call _ns_execute_primitive()
2960 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
2961 vnf_index
, seq
.get("name"))
2962 self
.logger
.debug(logging_text
+ step
)
2963 # Create the primitive for each sequence, i.e. "primitive": "touch"
2964 primitive
= seq
.get('name')
2965 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
2968 self
._add
_suboperation
(db_nslcmop
,
2974 mapped_primitive_params
)
2975 # Sub-operations: Call _ns_execute_primitive() instead of action()
2977 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
2978 mapped_primitive_params
,
2980 except LcmException
:
2981 # this happens when VCA is not deployed. In this case it is not needed to terminate
2983 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
2984 if result
not in result_ok
:
2985 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
2986 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
2987 # set that this VCA do not need terminated
2988 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
2989 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
2991 if vca_deployed
.get("prometheus_jobs") and self
.prometheus
:
2992 await self
.prometheus
.update(remove_jobs
=vca_deployed
["prometheus_jobs"])
2995 await self
.vca_map
[vca_type
].delete_execution_environment(vca_deployed
["ee_id"])
2997 async def _delete_all_N2VC(self
, db_nsr
: dict):
2998 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
2999 namespace
= "." + db_nsr
["_id"]
3001 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
3002 except N2VCNotFound
: # already deleted. Skip
3004 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
3006 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
3008 Terminates a deployment from RO
3009 :param logging_text:
3010 :param nsr_deployed: db_nsr._admin.deployed
3013 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
3014 this method will update only the index 2, but it will write on database the concatenated content of the list
3019 ro_nsr_id
= ro_delete_action
= None
3020 if nsr_deployed
and nsr_deployed
.get("RO"):
3021 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
3022 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
3025 stage
[2] = "Deleting ns from VIM."
3026 db_nsr_update
["detailed-status"] = " ".join(stage
)
3027 self
._write
_op
_status
(nslcmop_id
, stage
)
3028 self
.logger
.debug(logging_text
+ stage
[2])
3029 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3030 self
._write
_op
_status
(nslcmop_id
, stage
)
3031 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
3032 ro_delete_action
= desc
["action_id"]
3033 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
3034 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3035 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3036 if ro_delete_action
:
3037 # wait until NS is deleted from VIM
3038 stage
[2] = "Waiting ns deleted from VIM."
3039 detailed_status_old
= None
3040 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
3042 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3043 self
._write
_op
_status
(nslcmop_id
, stage
)
3045 delete_timeout
= 20 * 60 # 20 minutes
3046 while delete_timeout
> 0:
3047 desc
= await self
.RO
.show(
3049 item_id_name
=ro_nsr_id
,
3050 extra_item
="action",
3051 extra_item_id
=ro_delete_action
)
3054 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3056 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3057 if ns_status
== "ERROR":
3058 raise ROclient
.ROClientException(ns_status_info
)
3059 elif ns_status
== "BUILD":
3060 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
3061 elif ns_status
== "ACTIVE":
3062 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3063 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3066 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3067 if stage
[2] != detailed_status_old
:
3068 detailed_status_old
= stage
[2]
3069 db_nsr_update
["detailed-status"] = " ".join(stage
)
3070 self
._write
_op
_status
(nslcmop_id
, stage
)
3071 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3072 await asyncio
.sleep(5, loop
=self
.loop
)
3074 else: # delete_timeout <= 0:
3075 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
3077 except Exception as e
:
3078 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3079 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3080 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
3081 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
3082 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
3083 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
3084 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3085 failed_detail
.append("delete conflict: {}".format(e
))
3086 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
3088 failed_detail
.append("delete error: {}".format(e
))
3089 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
3092 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
3093 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
3095 stage
[2] = "Deleting nsd from RO."
3096 db_nsr_update
["detailed-status"] = " ".join(stage
)
3097 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3098 self
._write
_op
_status
(nslcmop_id
, stage
)
3099 await self
.RO
.delete("nsd", ro_nsd_id
)
3100 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
3101 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3102 except Exception as e
:
3103 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3104 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
3105 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
3106 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3107 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
3108 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3110 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
3111 self
.logger
.error(logging_text
+ failed_detail
[-1])
3113 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
3114 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
3115 if not vnf_deployed
or not vnf_deployed
["id"]:
3118 ro_vnfd_id
= vnf_deployed
["id"]
3119 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
3120 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
3121 db_nsr_update
["detailed-status"] = " ".join(stage
)
3122 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3123 self
._write
_op
_status
(nslcmop_id
, stage
)
3124 await self
.RO
.delete("vnfd", ro_vnfd_id
)
3125 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
3126 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3127 except Exception as e
:
3128 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
3129 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
3130 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
3131 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
3132 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
3133 self
.logger
.debug(logging_text
+ failed_detail
[-1])
3135 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
3136 self
.logger
.error(logging_text
+ failed_detail
[-1])
3139 stage
[2] = "Error deleting from VIM"
3141 stage
[2] = "Deleted from VIM"
3142 db_nsr_update
["detailed-status"] = " ".join(stage
)
3143 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3144 self
._write
_op
_status
(nslcmop_id
, stage
)
3147 raise LcmException("; ".join(failed_detail
))
3149 async def terminate(self
, nsr_id
, nslcmop_id
):
3150 # Try to lock HA task here
3151 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3152 if not task_is_locked_by_me
:
3155 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
3156 self
.logger
.debug(logging_text
+ "Enter")
3157 timeout_ns_terminate
= self
.timeout_ns_terminate
3160 operation_params
= None
3162 error_list
= [] # annotates all failed error messages
3163 db_nslcmop_update
= {}
3164 autoremove
= False # autoremove after terminated
3165 tasks_dict_info
= {}
3167 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
3168 # ^ contains [stage, step, VIM-status]
3170 # wait for any previous tasks in process
3171 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
3173 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
3174 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3175 operation_params
= db_nslcmop
.get("operationParams") or {}
3176 if operation_params
.get("timeout_ns_terminate"):
3177 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
3178 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
3179 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3181 db_nsr_update
["operational-status"] = "terminating"
3182 db_nsr_update
["config-status"] = "terminating"
3183 self
._write
_ns
_status
(
3185 ns_state
="TERMINATING",
3186 current_operation
="TERMINATING",
3187 current_operation_id
=nslcmop_id
,
3188 other_update
=db_nsr_update
3190 self
._write
_op
_status
(
3195 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
3196 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
3199 stage
[1] = "Getting vnf descriptors from db."
3200 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
3201 db_vnfds_from_id
= {}
3202 db_vnfds_from_member_index
= {}
3204 for vnfr
in db_vnfrs_list
:
3205 vnfd_id
= vnfr
["vnfd-id"]
3206 if vnfd_id
not in db_vnfds_from_id
:
3207 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
3208 db_vnfds_from_id
[vnfd_id
] = vnfd
3209 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
3211 # Destroy individual execution environments when there are terminating primitives.
3212 # Rest of EE will be deleted at once
3213 # TODO - check before calling _destroy_N2VC
3214 # if not operation_params.get("skip_terminate_primitives"):#
3215 # or not vca.get("needed_terminate"):
3216 stage
[0] = "Stage 2/3 execute terminating primitives."
3217 self
.logger
.debug(logging_text
+ stage
[0])
3218 stage
[1] = "Looking execution environment that needs terminate."
3219 self
.logger
.debug(logging_text
+ stage
[1])
3220 # self.logger.debug("nsr_deployed: {}".format(nsr_deployed))
3221 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
3222 config_descriptor
= None
3223 if not vca
or not vca
.get("ee_id"):
3225 if not vca
.get("member-vnf-index"):
3227 config_descriptor
= db_nsr
.get("ns-configuration")
3228 elif vca
.get("vdu_id"):
3229 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3230 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
3232 config_descriptor
= vdud
.get("vdu-configuration")
3233 elif vca
.get("kdu_name"):
3234 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3235 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
3237 config_descriptor
= kdud
.get("kdu-configuration")
3239 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
3240 vca_type
= vca
.get("type")
3241 exec_terminate_primitives
= (not operation_params
.get("skip_terminate_primitives") and
3242 vca
.get("needed_terminate"))
3243 # For helm we must destroy_ee. Also for native_charm, as juju_model cannot be deleted if there are
3244 # pending native charms
3245 destroy_ee
= True if vca_type
in ("helm", "native_charm") else False
3246 # self.logger.debug(logging_text + "vca_index: {}, ee_id: {}, vca_type: {} destroy_ee: {}".format(
3247 # vca_index, vca.get("ee_id"), vca_type, destroy_ee))
3248 task
= asyncio
.ensure_future(
3249 self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
, vca_index
,
3250 destroy_ee
, exec_terminate_primitives
))
3251 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
3253 # wait for pending tasks of terminate primitives
3255 self
.logger
.debug(logging_text
+ 'Waiting for tasks {}'.format(list(tasks_dict_info
.keys())))
3256 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
3257 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
3259 tasks_dict_info
.clear()
3261 return # raise LcmException("; ".join(error_list))
3263 # remove All execution environments at once
3264 stage
[0] = "Stage 3/3 delete all."
3266 if nsr_deployed
.get("VCA"):
3267 stage
[1] = "Deleting all execution environments."
3268 self
.logger
.debug(logging_text
+ stage
[1])
3269 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
3270 timeout
=self
.timeout_charm_delete
))
3271 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
3272 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
3274 # Delete from k8scluster
3275 stage
[1] = "Deleting KDUs."
3276 self
.logger
.debug(logging_text
+ stage
[1])
3277 # print(nsr_deployed)
3278 for kdu
in get_iterable(nsr_deployed
, "K8s"):
3279 if not kdu
or not kdu
.get("kdu-instance"):
3281 kdu_instance
= kdu
.get("kdu-instance")
3282 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
3283 task_delete_kdu_instance
= asyncio
.ensure_future(
3284 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
3285 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3286 kdu_instance
=kdu_instance
))
3288 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
3289 format(kdu
.get("k8scluster-type")))
3291 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
3294 stage
[1] = "Deleting ns from VIM."
3296 task_delete_ro
= asyncio
.ensure_future(
3297 self
._terminate
_ng
_ro
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3299 task_delete_ro
= asyncio
.ensure_future(
3300 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3301 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
3303 # rest of staff will be done at finally
3305 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
3306 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3308 except asyncio
.CancelledError
:
3309 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
3310 exc
= "Operation was cancelled"
3311 except Exception as e
:
3312 exc
= traceback
.format_exc()
3313 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
3316 error_list
.append(str(exc
))
3318 # wait for pending tasks
3320 stage
[1] = "Waiting for terminate pending tasks."
3321 self
.logger
.debug(logging_text
+ stage
[1])
3322 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
3324 stage
[1] = stage
[2] = ""
3325 except asyncio
.CancelledError
:
3326 error_list
.append("Cancelled")
3327 # TODO cancell all tasks
3328 except Exception as exc
:
3329 error_list
.append(str(exc
))
3330 # update status at database
3332 error_detail
= "; ".join(error_list
)
3333 # self.logger.error(logging_text + error_detail)
3334 error_description_nslcmop
= '{} Detail: {}'.format(stage
[0], error_detail
)
3335 error_description_nsr
= 'Operation: TERMINATING.{}, {}.'.format(nslcmop_id
, stage
[0])
3337 db_nsr_update
["operational-status"] = "failed"
3338 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
3339 db_nslcmop_update
["detailed-status"] = error_detail
3340 nslcmop_operation_state
= "FAILED"
3344 error_description_nsr
= error_description_nslcmop
= None
3345 ns_state
= "NOT_INSTANTIATED"
3346 db_nsr_update
["operational-status"] = "terminated"
3347 db_nsr_update
["detailed-status"] = "Done"
3348 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
3349 db_nslcmop_update
["detailed-status"] = "Done"
3350 nslcmop_operation_state
= "COMPLETED"
3353 self
._write
_ns
_status
(
3356 current_operation
="IDLE",
3357 current_operation_id
=None,
3358 error_description
=error_description_nsr
,
3359 error_detail
=error_detail
,
3360 other_update
=db_nsr_update
3362 self
._write
_op
_status
(
3365 error_message
=error_description_nslcmop
,
3366 operation_state
=nslcmop_operation_state
,
3367 other_update
=db_nslcmop_update
,
3369 if ns_state
== "NOT_INSTANTIATED":
3371 self
.db
.set_list("vnfrs", {"nsr-id-ref": nsr_id
}, {"_admin.nsState": "NOT_INSTANTIATED"})
3372 except DbException
as e
:
3373 self
.logger
.warn(logging_text
+ 'Error writing VNFR status for nsr-id-ref: {} -> {}'.
3375 if operation_params
:
3376 autoremove
= operation_params
.get("autoremove", False)
3377 if nslcmop_operation_state
:
3379 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3380 "operationState": nslcmop_operation_state
,
3381 "autoremove": autoremove
},
3383 except Exception as e
:
3384 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3386 self
.logger
.debug(logging_text
+ "Exit")
3387 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
3389 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
3391 error_detail_list
= []
3393 pending_tasks
= list(created_tasks_info
.keys())
3394 num_tasks
= len(pending_tasks
)
3396 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3397 self
._write
_op
_status
(nslcmop_id
, stage
)
3398 while pending_tasks
:
3400 _timeout
= timeout
+ time_start
- time()
3401 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
3402 return_when
=asyncio
.FIRST_COMPLETED
)
3403 num_done
+= len(done
)
3404 if not done
: # Timeout
3405 for task
in pending_tasks
:
3406 new_error
= created_tasks_info
[task
] + ": Timeout"
3407 error_detail_list
.append(new_error
)
3408 error_list
.append(new_error
)
3411 if task
.cancelled():
3414 exc
= task
.exception()
3416 if isinstance(exc
, asyncio
.TimeoutError
):
3418 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
3419 error_list
.append(created_tasks_info
[task
])
3420 error_detail_list
.append(new_error
)
3421 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
3423 self
.logger
.error(logging_text
+ new_error
)
3425 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
3426 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
3428 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
3429 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3431 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
3432 if nsr_id
: # update also nsr
3433 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
3434 "errorDetail": ". ".join(error_detail_list
)})
3435 self
._write
_op
_status
(nslcmop_id
, stage
)
3436 return error_detail_list
3439 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
3441 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
3442 The default-value is used. If it is between < > it look for a value at instantiation_params
3443 :param primitive_desc: portion of VNFD/NSD that describes primitive
3444 :param params: Params provided by user
3445 :param instantiation_params: Instantiation params provided by user
3446 :return: a dictionary with the calculated params
3448 calculated_params
= {}
3449 for parameter
in primitive_desc
.get("parameter", ()):
3450 param_name
= parameter
["name"]
3451 if param_name
in params
:
3452 calculated_params
[param_name
] = params
[param_name
]
3453 elif "default-value" in parameter
or "value" in parameter
:
3454 if "value" in parameter
:
3455 calculated_params
[param_name
] = parameter
["value"]
3457 calculated_params
[param_name
] = parameter
["default-value"]
3458 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3459 and calculated_params
[param_name
].endswith(">"):
3460 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3461 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3463 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3464 format(calculated_params
[param_name
], primitive_desc
["name"]))
3466 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3467 format(param_name
, primitive_desc
["name"]))
3469 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3470 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3472 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3473 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3475 # add always ns_config_info if primitive name is config
3476 if primitive_desc
["name"] == "config":
3477 if "ns_config_info" in instantiation_params
:
3478 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3479 return calculated_params
3481 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None,
3482 ee_descriptor_id
=None):
3483 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3484 for vca
in deployed_vca
:
3487 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3489 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3491 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3493 if ee_descriptor_id
and ee_descriptor_id
!= vca
["ee_descriptor_id"]:
3497 # vca_deployed not found
3498 raise LcmException("charm for member_vnf_index={} vdu_id={}.{} kdu_name={} execution-environment-list.id={}"
3499 " is not deployed".format(member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
,
3503 ee_id
= vca
.get("ee_id")
3504 vca_type
= vca
.get("type", "lxc_proxy_charm") # default value for backward compatibility - proxy charm
3506 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3507 "execution environment"
3508 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3509 return ee_id
, vca_type
3511 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3512 retries_interval
=30, timeout
=None,
3513 vca_type
=None, db_dict
=None) -> (str, str):
3515 if primitive
== "config":
3516 primitive_params
= {"params": primitive_params
}
3518 vca_type
= vca_type
or "lxc_proxy_charm"
3522 output
= await asyncio
.wait_for(
3523 self
.vca_map
[vca_type
].exec_primitive(
3525 primitive_name
=primitive
,
3526 params_dict
=primitive_params
,
3527 progress_timeout
=self
.timeout_progress_primitive
,
3528 total_timeout
=self
.timeout_primitive
,
3530 timeout
=timeout
or self
.timeout_primitive
)
3533 except asyncio
.CancelledError
:
3535 except Exception as e
: # asyncio.TimeoutError
3536 if isinstance(e
, asyncio
.TimeoutError
):
3540 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3542 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3544 return 'FAILED', str(e
)
3546 return 'COMPLETED', output
3548 except (LcmException
, asyncio
.CancelledError
):
3550 except Exception as e
:
3551 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3553 async def action(self
, nsr_id
, nslcmop_id
):
3555 # Try to lock HA task here
3556 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3557 if not task_is_locked_by_me
:
3560 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3561 self
.logger
.debug(logging_text
+ "Enter")
3562 # get all needed from database
3566 db_nslcmop_update
= {}
3567 nslcmop_operation_state
= None
3568 error_description_nslcmop
= None
3571 # wait for any previous tasks in process
3572 step
= "Waiting for previous operations to terminate"
3573 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3575 self
._write
_ns
_status
(
3578 current_operation
="RUNNING ACTION",
3579 current_operation_id
=nslcmop_id
3582 step
= "Getting information from database"
3583 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3584 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3586 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3587 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3588 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3589 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3590 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3591 primitive
= db_nslcmop
["operationParams"]["primitive"]
3592 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3593 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3596 step
= "Getting vnfr from database"
3597 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3598 step
= "Getting vnfd from database"
3599 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3601 step
= "Getting nsd from database"
3602 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3604 # for backward compatibility
3605 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3606 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3607 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3608 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3610 # look for primitive
3611 config_primitive_desc
= descriptor_configuration
= None
3613 for vdu
in get_iterable(db_vnfd
, "vdu"):
3614 if vdu_id
== vdu
["id"]:
3615 descriptor_configuration
= vdu
.get("vdu-configuration")
3618 for kdu
in get_iterable(db_vnfd
, "kdu"):
3619 if kdu_name
== kdu
["name"]:
3620 descriptor_configuration
= kdu
.get("kdu-configuration")
3623 descriptor_configuration
= db_vnfd
.get("vnf-configuration")
3625 descriptor_configuration
= db_nsd
.get("ns-configuration")
3627 if descriptor_configuration
and descriptor_configuration
.get("config-primitive"):
3628 for config_primitive
in descriptor_configuration
["config-primitive"]:
3629 if config_primitive
["name"] == primitive
:
3630 config_primitive_desc
= config_primitive
3633 if not config_primitive_desc
:
3634 if not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3635 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3637 primitive_name
= primitive
3638 ee_descriptor_id
= None
3640 primitive_name
= config_primitive_desc
.get("execution-environment-primitive", primitive
)
3641 ee_descriptor_id
= config_primitive_desc
.get("execution-environment-ref")
3645 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3646 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3648 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3649 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3651 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3653 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3656 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3658 # TODO check if ns is in a proper status
3659 if kdu_name
and (primitive_name
in ("upgrade", "rollback", "status") or kdu_action
):
3660 # kdur and desc_params already set from before
3661 if primitive_params
:
3662 desc_params
.update(primitive_params
)
3663 # TODO Check if we will need something at vnf level
3664 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3665 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3668 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3670 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3671 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3672 raise LcmException(msg
)
3674 db_dict
= {"collection": "nsrs",
3675 "filter": {"_id": nsr_id
},
3676 "path": "_admin.deployed.K8s.{}".format(index
)}
3677 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive_name
, vnf_index
, kdu_name
))
3678 step
= "Executing kdu {}".format(primitive_name
)
3679 if primitive_name
== "upgrade":
3680 if desc_params
.get("kdu_model"):
3681 kdu_model
= desc_params
.get("kdu_model")
3682 del desc_params
["kdu_model"]
3684 kdu_model
= kdu
.get("kdu-model")
3685 parts
= kdu_model
.split(sep
=":")
3687 kdu_model
= parts
[0]
3689 detailed_status
= await asyncio
.wait_for(
3690 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3691 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3692 kdu_instance
=kdu
.get("kdu-instance"),
3693 atomic
=True, kdu_model
=kdu_model
,
3694 params
=desc_params
, db_dict
=db_dict
,
3695 timeout
=timeout_ns_action
),
3696 timeout
=timeout_ns_action
+ 10)
3697 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3698 elif primitive_name
== "rollback":
3699 detailed_status
= await asyncio
.wait_for(
3700 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3701 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3702 kdu_instance
=kdu
.get("kdu-instance"),
3704 timeout
=timeout_ns_action
)
3705 elif primitive_name
== "status":
3706 detailed_status
= await asyncio
.wait_for(
3707 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3708 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3709 kdu_instance
=kdu
.get("kdu-instance")),
3710 timeout
=timeout_ns_action
)
3712 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3713 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3715 detailed_status
= await asyncio
.wait_for(
3716 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3717 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3718 kdu_instance
=kdu_instance
,
3719 primitive_name
=primitive_name
,
3720 params
=params
, db_dict
=db_dict
,
3721 timeout
=timeout_ns_action
),
3722 timeout
=timeout_ns_action
)
3725 nslcmop_operation_state
= 'COMPLETED'
3727 detailed_status
= ''
3728 nslcmop_operation_state
= 'FAILED'
3730 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3731 member_vnf_index
=vnf_index
,
3733 vdu_count_index
=vdu_count_index
,
3734 ee_descriptor_id
=ee_descriptor_id
)
3735 db_nslcmop_notif
= {"collection": "nslcmops",
3736 "filter": {"_id": nslcmop_id
},
3737 "path": "admin.VCA"}
3738 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3740 primitive
=primitive_name
,
3741 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3742 timeout
=timeout_ns_action
,
3744 db_dict
=db_nslcmop_notif
)
3746 db_nslcmop_update
["detailed-status"] = detailed_status
3747 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3748 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3750 return # database update is called inside finally
3752 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3753 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3755 except asyncio
.CancelledError
:
3756 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3757 exc
= "Operation was cancelled"
3758 except asyncio
.TimeoutError
:
3759 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3761 except Exception as e
:
3762 exc
= traceback
.format_exc()
3763 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3766 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3767 "FAILED {}: {}".format(step
, exc
)
3768 nslcmop_operation_state
= "FAILED"
3770 self
._write
_ns
_status
(
3772 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3773 current_operation
="IDLE",
3774 current_operation_id
=None,
3775 # error_description=error_description_nsr,
3776 # error_detail=error_detail,
3777 other_update
=db_nsr_update
3780 self
._write
_op
_status
(
3783 error_message
=error_description_nslcmop
,
3784 operation_state
=nslcmop_operation_state
,
3785 other_update
=db_nslcmop_update
,
3788 if nslcmop_operation_state
:
3790 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3791 "operationState": nslcmop_operation_state
},
3793 except Exception as e
:
3794 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3795 self
.logger
.debug(logging_text
+ "Exit")
3796 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3797 return nslcmop_operation_state
, detailed_status
3799 async def scale(self
, nsr_id
, nslcmop_id
):
3801 # Try to lock HA task here
3802 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3803 if not task_is_locked_by_me
:
3806 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3807 self
.logger
.debug(logging_text
+ "Enter")
3808 # get all needed from database
3811 db_nslcmop_update
= {}
3812 nslcmop_operation_state
= None
3815 # in case of error, indicates what part of scale was failed to put nsr at error status
3816 scale_process
= None
3817 old_operational_status
= ""
3818 old_config_status
= ""
3821 # wait for any previous tasks in process
3822 step
= "Waiting for previous operations to terminate"
3823 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3825 self
._write
_ns
_status
(
3828 current_operation
="SCALING",
3829 current_operation_id
=nslcmop_id
3832 step
= "Getting nslcmop from database"
3833 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3834 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3835 step
= "Getting nsr from database"
3836 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3838 old_operational_status
= db_nsr
["operational-status"]
3839 old_config_status
= db_nsr
["config-status"]
3840 step
= "Parsing scaling parameters"
3841 # self.logger.debug(step)
3842 db_nsr_update
["operational-status"] = "scaling"
3843 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3844 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3847 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3848 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3849 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3850 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3851 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3854 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3855 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3856 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3857 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3858 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3860 # for backward compatibility
3861 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3862 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3863 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3864 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3866 step
= "Getting vnfr from database"
3867 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3868 step
= "Getting vnfd from database"
3869 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3871 step
= "Getting scaling-group-descriptor"
3872 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3873 if scaling_descriptor
["name"] == scaling_group
:
3876 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3877 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3880 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3881 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3882 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3885 # TODO check if ns is in a proper status
3886 step
= "Sending scale order to VIM"
3888 if not db_nsr
["_admin"].get("scaling-group"):
3889 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3890 admin_scale_index
= 0
3892 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3893 if admin_scale_info
["name"] == scaling_group
:
3894 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3896 else: # not found, set index one plus last element and add new entry with the name
3897 admin_scale_index
+= 1
3898 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3899 RO_scaling_info
= []
3900 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3901 if scaling_type
== "SCALE_OUT":
3902 # count if max-instance-count is reached
3903 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3904 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3905 if nb_scale_op
>= max_instance_count
:
3906 raise LcmException("reached the limit of {} (max-instance-count) "
3907 "scaling-out operations for the "
3908 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3911 vdu_scaling_info
["scaling_direction"] = "OUT"
3912 vdu_scaling_info
["vdu-create"] = {}
3913 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3914 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3915 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3916 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3918 elif scaling_type
== "SCALE_IN":
3919 # count if min-instance-count is reached
3920 min_instance_count
= 0
3921 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3922 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3923 if nb_scale_op
<= min_instance_count
:
3924 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3925 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3927 vdu_scaling_info
["scaling_direction"] = "IN"
3928 vdu_scaling_info
["vdu-delete"] = {}
3929 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3930 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3931 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3932 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3934 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3935 vdu_create
= vdu_scaling_info
.get("vdu-create")
3936 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3937 if vdu_scaling_info
["scaling_direction"] == "IN":
3938 for vdur
in reversed(db_vnfr
["vdur"]):
3939 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3940 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3941 vdu_scaling_info
["vdu"].append({
3942 "name": vdur
["name"],
3943 "vdu_id": vdur
["vdu-id-ref"],
3946 for interface
in vdur
["interfaces"]:
3947 vdu_scaling_info
["vdu"][-1]["interface"].append({
3948 "name": interface
["name"],
3949 "ip_address": interface
["ip-address"],
3950 "mac_address": interface
.get("mac-address"),
3952 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
3955 step
= "Executing pre-scale vnf-config-primitive"
3956 if scaling_descriptor
.get("scaling-config-action"):
3957 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3958 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
3959 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
3960 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3961 step
= db_nslcmop_update
["detailed-status"] = \
3962 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3964 # look for primitive
3965 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3966 if config_primitive
["name"] == vnf_config_primitive
:
3970 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
3971 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
3972 "primitive".format(scaling_group
, vnf_config_primitive
))
3974 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3975 if db_vnfr
.get("additionalParamsForVnf"):
3976 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3978 scale_process
= "VCA"
3979 db_nsr_update
["config-status"] = "configuring pre-scaling"
3980 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3982 # Pre-scale retry check: Check if this sub-operation has been executed before
3983 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3984 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
3985 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3986 # Skip sub-operation
3987 result
= 'COMPLETED'
3988 result_detail
= 'Done'
3989 self
.logger
.debug(logging_text
+
3990 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
3991 vnf_config_primitive
, result
, result_detail
))
3993 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3994 # New sub-operation: Get index of this sub-operation
3995 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3996 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3997 format(vnf_config_primitive
))
3999 # retry: Get registered params for this existing sub-operation
4000 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4001 vnf_index
= op
.get('member_vnf_index')
4002 vnf_config_primitive
= op
.get('primitive')
4003 primitive_params
= op
.get('primitive_params')
4004 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
4005 format(vnf_config_primitive
))
4006 # Execute the primitive, either with new (first-time) or registered (reintent) args
4007 ee_descriptor_id
= config_primitive
.get("execution-environment-ref")
4008 primitive_name
= config_primitive
.get("execution-environment-primitive",
4009 vnf_config_primitive
)
4010 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
4011 member_vnf_index
=vnf_index
,
4013 vdu_count_index
=None,
4014 ee_descriptor_id
=ee_descriptor_id
)
4015 result
, result_detail
= await self
._ns
_execute
_primitive
(
4016 ee_id
, primitive_name
, primitive_params
, vca_type
)
4017 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
4018 vnf_config_primitive
, result
, result_detail
))
4019 # Update operationState = COMPLETED | FAILED
4020 self
._update
_suboperation
_status
(
4021 db_nslcmop
, op_index
, result
, result_detail
)
4023 if result
== "FAILED":
4024 raise LcmException(result_detail
)
4025 db_nsr_update
["config-status"] = old_config_status
4026 scale_process
= None
4030 # Should this block be skipped if 'RO_nsr_id' == None ?
4031 # if (RO_nsr_id and RO_scaling_info):
4033 scale_process
= "RO"
4034 # Scale RO retry check: Check if this sub-operation has been executed before
4035 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4036 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
4037 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4038 # Skip sub-operation
4039 result
= 'COMPLETED'
4040 result_detail
= 'Done'
4041 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
4042 result
, result_detail
))
4044 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4045 # New sub-operation: Get index of this sub-operation
4046 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4047 self
.logger
.debug(logging_text
+ "New sub-operation RO")
4049 # retry: Get registered params for this existing sub-operation
4050 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4051 RO_nsr_id
= op
.get('RO_nsr_id')
4052 RO_scaling_info
= op
.get('RO_scaling_info')
4053 self
.logger
.debug(logging_text
+ "Sub-operation RO retry for primitive {}".format(
4054 vnf_config_primitive
))
4056 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
4057 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
4058 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
4060 RO_nslcmop_id
= RO_desc
["instance_action_id"]
4061 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
4063 RO_task_done
= False
4064 step
= detailed_status
= "Waiting RO_task_id={} to complete the scale action.".format(RO_nslcmop_id
)
4065 detailed_status_old
= None
4066 self
.logger
.debug(logging_text
+ step
)
4068 deployment_timeout
= 1 * 3600 # One hour
4069 while deployment_timeout
> 0:
4070 if not RO_task_done
:
4071 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
4072 extra_item_id
=RO_nslcmop_id
)
4075 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4077 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
4078 if ns_status
== "ERROR":
4079 raise ROclient
.ROClientException(ns_status_info
)
4080 elif ns_status
== "BUILD":
4081 detailed_status
= step
+ "; {}".format(ns_status_info
)
4082 elif ns_status
== "ACTIVE":
4084 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
4085 self
.logger
.debug(logging_text
+ step
)
4087 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
4090 if ns_status
== "ERROR":
4091 raise ROclient
.ROClientException(ns_status_info
)
4092 elif ns_status
== "BUILD":
4093 detailed_status
= step
+ "; {}".format(ns_status_info
)
4094 elif ns_status
== "ACTIVE":
4095 step
= detailed_status
= \
4096 "Waiting for management IP address reported by the VIM. Updating VNFRs"
4098 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
4101 desc
= await self
.RO
.show("ns", RO_nsr_id
)
4104 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
4106 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
4107 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
4109 except LcmExceptionNoMgmtIP
:
4112 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
4113 if detailed_status
!= detailed_status_old
:
4114 self
._update
_suboperation
_status
(
4115 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
4116 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
4117 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
4119 await asyncio
.sleep(5, loop
=self
.loop
)
4120 deployment_timeout
-= 5
4121 if deployment_timeout
<= 0:
4122 self
._update
_suboperation
_status
(
4123 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
4124 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
4126 # update VDU_SCALING_INFO with the obtained ip_addresses
4127 if vdu_scaling_info
["scaling_direction"] == "OUT":
4128 for vdur
in reversed(db_vnfr
["vdur"]):
4129 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
4130 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
4131 vdu_scaling_info
["vdu"].append({
4132 "name": vdur
["name"],
4133 "vdu_id": vdur
["vdu-id-ref"],
4136 for interface
in vdur
["interfaces"]:
4137 vdu_scaling_info
["vdu"][-1]["interface"].append({
4138 "name": interface
["name"],
4139 "ip_address": interface
["ip-address"],
4140 "mac_address": interface
.get("mac-address"),
4142 del vdu_scaling_info
["vdu-create"]
4144 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
4147 scale_process
= None
4149 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
4152 # execute primitive service POST-SCALING
4153 step
= "Executing post-scale vnf-config-primitive"
4154 if scaling_descriptor
.get("scaling-config-action"):
4155 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
4156 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
4157 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
4158 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
4159 step
= db_nslcmop_update
["detailed-status"] = \
4160 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
4162 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
4163 if db_vnfr
.get("additionalParamsForVnf"):
4164 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
4166 # look for primitive
4167 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
4168 if config_primitive
["name"] == vnf_config_primitive
:
4172 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-"
4173 "action[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:"
4174 "config-primitive".format(scaling_group
, vnf_config_primitive
))
4175 scale_process
= "VCA"
4176 db_nsr_update
["config-status"] = "configuring post-scaling"
4177 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
4179 # Post-scale retry check: Check if this sub-operation has been executed before
4180 op_index
= self
._check
_or
_add
_scale
_suboperation
(
4181 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
4182 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
4183 # Skip sub-operation
4184 result
= 'COMPLETED'
4185 result_detail
= 'Done'
4186 self
.logger
.debug(logging_text
+
4187 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
4188 format(vnf_config_primitive
, result
, result_detail
))
4190 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
4191 # New sub-operation: Get index of this sub-operation
4192 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
4193 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
4194 format(vnf_config_primitive
))
4196 # retry: Get registered params for this existing sub-operation
4197 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
4198 vnf_index
= op
.get('member_vnf_index')
4199 vnf_config_primitive
= op
.get('primitive')
4200 primitive_params
= op
.get('primitive_params')
4201 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
4202 format(vnf_config_primitive
))
4203 # Execute the primitive, either with new (first-time) or registered (reintent) args
4204 ee_descriptor_id
= config_primitive
.get("execution-environment-ref")
4205 primitive_name
= config_primitive
.get("execution-environment-primitive",
4206 vnf_config_primitive
)
4207 ee_id
, vca_type
= self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
4208 member_vnf_index
=vnf_index
,
4210 vdu_count_index
=None,
4211 ee_descriptor_id
=ee_descriptor_id
)
4212 result
, result_detail
= await self
._ns
_execute
_primitive
(
4213 ee_id
, primitive_name
, primitive_params
, vca_type
)
4214 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
4215 vnf_config_primitive
, result
, result_detail
))
4216 # Update operationState = COMPLETED | FAILED
4217 self
._update
_suboperation
_status
(
4218 db_nslcmop
, op_index
, result
, result_detail
)
4220 if result
== "FAILED":
4221 raise LcmException(result_detail
)
4222 db_nsr_update
["config-status"] = old_config_status
4223 scale_process
= None
4226 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
4227 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
4228 else old_operational_status
4229 db_nsr_update
["config-status"] = old_config_status
4231 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
4232 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4234 except asyncio
.CancelledError
:
4235 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
4236 exc
= "Operation was cancelled"
4237 except Exception as e
:
4238 exc
= traceback
.format_exc()
4239 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
4241 self
._write
_ns
_status
(
4244 current_operation
="IDLE",
4245 current_operation_id
=None
4248 db_nslcmop_update
["detailed-status"] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
4249 nslcmop_operation_state
= "FAILED"
4251 db_nsr_update
["operational-status"] = old_operational_status
4252 db_nsr_update
["config-status"] = old_config_status
4253 db_nsr_update
["detailed-status"] = ""
4255 if "VCA" in scale_process
:
4256 db_nsr_update
["config-status"] = "failed"
4257 if "RO" in scale_process
:
4258 db_nsr_update
["operational-status"] = "failed"
4259 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
4262 error_description_nslcmop
= None
4263 nslcmop_operation_state
= "COMPLETED"
4264 db_nslcmop_update
["detailed-status"] = "Done"
4266 self
._write
_op
_status
(
4269 error_message
=error_description_nslcmop
,
4270 operation_state
=nslcmop_operation_state
,
4271 other_update
=db_nslcmop_update
,
4274 self
._write
_ns
_status
(
4277 current_operation
="IDLE",
4278 current_operation_id
=None,
4279 other_update
=db_nsr_update
4282 if nslcmop_operation_state
:
4284 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
4285 "operationState": nslcmop_operation_state
},
4288 # await asyncio.sleep(cooldown_time, loop=self.loop)
4289 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
4290 except Exception as e
:
4291 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
4292 self
.logger
.debug(logging_text
+ "Exit")
4293 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")
4295 async def add_prometheus_metrics(self
, ee_id
, artifact_path
, ee_config_descriptor
, vnfr_id
, nsr_id
, target_ip
):
4296 if not self
.prometheus
:
4298 # look if exist a file called 'prometheus*.j2' and
4299 artifact_content
= self
.fs
.dir_ls(artifact_path
)
4300 job_file
= next((f
for f
in artifact_content
if f
.startswith("prometheus") and f
.endswith(".j2")), None)
4303 with self
.fs
.file_open((artifact_path
, job_file
), "r") as f
:
4307 _
, _
, service
= ee_id
.partition(".") # remove prefix "namespace."
4308 host_name
= "{}-{}".format(service
, ee_config_descriptor
["metric-service"])
4310 vnfr_id
= vnfr_id
.replace("-", "")
4312 "JOB_NAME": vnfr_id
,
4313 "TARGET_IP": target_ip
,
4314 "EXPORTER_POD_IP": host_name
,
4315 "EXPORTER_POD_PORT": host_port
,
4317 job_list
= self
.prometheus
.parse_job(job_data
, variables
)
4318 # ensure job_name is using the vnfr_id. Adding the metadata nsr_id
4319 for job
in job_list
:
4320 if not isinstance(job
.get("job_name"), str) or vnfr_id
not in job
["job_name"]:
4321 job
["job_name"] = vnfr_id
+ "_" + str(randint(1, 10000))
4322 job
["nsr_id"] = nsr_id
4323 job_dict
= {jl
["job_name"]: jl
for jl
in job_list
}
4324 if await self
.prometheus
.update(job_dict
):
4325 return list(job_dict
.keys())