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
.lcm_utils
import LcmException
, LcmExceptionNoMgmtIP
, LcmBase
, deep_get
, get_iterable
, populate_dict
29 from n2vc
.k8s_helm_conn
import K8sHelmConnector
30 from n2vc
.k8s_juju_conn
import K8sJujuConnector
32 from osm_common
.dbbase
import DbException
33 from osm_common
.fsbase
import FsException
35 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
36 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
38 from copy
import copy
, deepcopy
39 from http
import HTTPStatus
41 from uuid
import uuid4
42 from functools
import partial
44 __author__
= "Alfonso Tierno"
48 timeout_vca_on_error
= 5 * 60 # Time for charm from first time at blocked,error status to mark as failed
49 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
50 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
51 timeout_charm_delete
= 10 * 60
52 timeout_primitive
= 10 * 60 # timeout for primitive execution
53 timeout_progress_primitive
= 2 * 60 # timeout for some progress in a primitive execution
55 SUBOPERATION_STATUS_NOT_FOUND
= -1
56 SUBOPERATION_STATUS_NEW
= -2
57 SUBOPERATION_STATUS_SKIP
= -3
58 task_name_deploy_vca
= "Deploying VCA"
60 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
62 Init, Connect to database, filesystem storage, and messaging
63 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
70 logger
=logging
.getLogger('lcm.ns')
74 self
.lcm_tasks
= lcm_tasks
75 self
.timeout
= config
["timeout"]
76 self
.ro_config
= config
["ro_config"]
77 self
.vca_config
= config
["VCA"].copy()
79 # create N2VC connector
80 self
.n2vc
= N2VCJujuConnector(
85 url
='{}:{}'.format(self
.vca_config
['host'], self
.vca_config
['port']),
86 username
=self
.vca_config
.get('user', None),
87 vca_config
=self
.vca_config
,
88 on_update_db
=self
._on
_update
_n
2vc
_db
91 self
.k8sclusterhelm
= K8sHelmConnector(
92 kubectl_command
=self
.vca_config
.get("kubectlpath"),
93 helm_command
=self
.vca_config
.get("helmpath"),
100 self
.k8sclusterjuju
= K8sJujuConnector(
101 kubectl_command
=self
.vca_config
.get("kubectlpath"),
102 juju_command
=self
.vca_config
.get("jujupath"),
109 self
.k8scluster_map
= {
110 "helm-chart": self
.k8sclusterhelm
,
111 "chart": self
.k8sclusterhelm
,
112 "juju-bundle": self
.k8sclusterjuju
,
113 "juju": self
.k8sclusterjuju
,
116 self
.RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
118 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
120 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
123 # TODO filter RO descriptor fields...
127 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
128 db_dict
['deploymentStatus'] = ro_descriptor
129 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
131 except Exception as e
:
132 self
.logger
.warn('Cannot write database RO deployment for ns={} -> {}'.format(nsrs_id
, e
))
134 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
):
136 # remove last dot from path (if exists)
137 if path
.endswith('.'):
140 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
141 # .format(table, filter, path, updated_data))
145 nsr_id
= filter.get('_id')
147 # read ns record from database
148 nsr
= self
.db
.get_one(table
='nsrs', q_filter
=filter)
149 current_ns_status
= nsr
.get('nsState')
151 # get vca status for NS
152 status_dict
= await self
.n2vc
.get_status(namespace
='.' + nsr_id
, yaml_format
=False)
156 db_dict
['vcaStatus'] = status_dict
158 # update configurationStatus for this VCA
160 vca_index
= int(path
[path
.rfind(".")+1:])
162 vca_list
= deep_get(target_dict
=nsr
, key_list
=('_admin', 'deployed', 'VCA'))
163 vca_status
= vca_list
[vca_index
].get('status')
165 configuration_status_list
= nsr
.get('configurationStatus')
166 config_status
= configuration_status_list
[vca_index
].get('status')
168 if config_status
== 'BROKEN' and vca_status
!= 'failed':
169 db_dict
['configurationStatus'][vca_index
] = 'READY'
170 elif config_status
!= 'BROKEN' and vca_status
== 'failed':
171 db_dict
['configurationStatus'][vca_index
] = 'BROKEN'
172 except Exception as e
:
173 # not update configurationStatus
174 self
.logger
.debug('Error updating vca_index (ignore): {}'.format(e
))
176 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
177 # if nsState = 'DEGRADED' check if all is OK
179 if current_ns_status
in ('READY', 'DEGRADED'):
180 error_description
= ''
182 if status_dict
.get('machines'):
183 for machine_id
in status_dict
.get('machines'):
184 machine
= status_dict
.get('machines').get(machine_id
)
185 # check machine agent-status
186 if machine
.get('agent-status'):
187 s
= machine
.get('agent-status').get('status')
190 error_description
+= 'machine {} agent-status={} ; '.format(machine_id
, s
)
191 # check machine instance status
192 if machine
.get('instance-status'):
193 s
= machine
.get('instance-status').get('status')
196 error_description
+= 'machine {} instance-status={} ; '.format(machine_id
, s
)
198 if status_dict
.get('applications'):
199 for app_id
in status_dict
.get('applications'):
200 app
= status_dict
.get('applications').get(app_id
)
201 # check application status
202 if app
.get('status'):
203 s
= app
.get('status').get('status')
206 error_description
+= 'application {} status={} ; '.format(app_id
, s
)
208 if error_description
:
209 db_dict
['errorDescription'] = error_description
210 if current_ns_status
== 'READY' and is_degraded
:
211 db_dict
['nsState'] = 'DEGRADED'
212 if current_ns_status
== 'DEGRADED' and not is_degraded
:
213 db_dict
['nsState'] = 'READY'
216 self
.update_db_2("nsrs", nsr_id
, db_dict
)
218 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
220 except Exception as e
:
221 self
.logger
.warn('Error updating NS state for ns={}: {}'.format(nsr_id
, e
))
223 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
225 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
226 :param vnfd: input vnfd
227 :param new_id: overrides vnf id if provided
228 :param additionalParams: Instantiation params for VNFs provided
229 :param nsrId: Id of the NSR
230 :return: copy of vnfd
233 vnfd_RO
= deepcopy(vnfd
)
234 # remove unused by RO configuration, monitoring, scaling and internal keys
235 vnfd_RO
.pop("_id", None)
236 vnfd_RO
.pop("_admin", None)
237 vnfd_RO
.pop("vnf-configuration", None)
238 vnfd_RO
.pop("monitoring-param", None)
239 vnfd_RO
.pop("scaling-group-descriptor", None)
240 vnfd_RO
.pop("kdu", None)
241 vnfd_RO
.pop("k8s-cluster", None)
243 vnfd_RO
["id"] = new_id
245 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
246 for vdu
in get_iterable(vnfd_RO
, "vdu"):
247 cloud_init_file
= None
248 if vdu
.get("cloud-init-file"):
249 base_folder
= vnfd
["_admin"]["storage"]
250 cloud_init_file
= "{}/{}/cloud_init/{}".format(base_folder
["folder"], base_folder
["pkg-dir"],
251 vdu
["cloud-init-file"])
252 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
253 cloud_init_content
= ci_file
.read()
254 vdu
.pop("cloud-init-file", None)
255 elif vdu
.get("cloud-init"):
256 cloud_init_content
= vdu
["cloud-init"]
261 ast
= env
.parse(cloud_init_content
)
262 mandatory_vars
= meta
.find_undeclared_variables(ast
)
264 for var
in mandatory_vars
:
265 if not additionalParams
or var
not in additionalParams
.keys():
266 raise LcmException("Variable '{}' defined at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
267 "file, must be provided in the instantiation parameters inside the "
268 "'additionalParamsForVnf' block".format(var
, vnfd
["id"], vdu
["id"]))
269 template
= Template(cloud_init_content
)
270 cloud_init_content
= template
.render(additionalParams
or {})
271 vdu
["cloud-init"] = cloud_init_content
274 except FsException
as e
:
275 raise LcmException("Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".
276 format(vnfd
["id"], vdu
["id"], cloud_init_file
, e
))
277 except (TemplateError
, TemplateNotFound
, TemplateSyntaxError
) as e
:
278 raise LcmException("Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".
279 format(vnfd
["id"], vdu
["id"], e
))
281 def _ns_params_2_RO(self
, ns_params
, nsd
, vnfd_dict
, n2vc_key_list
):
283 Creates a RO ns descriptor from OSM ns_instantiate params
284 :param ns_params: OSM instantiate params
285 :return: The RO ns descriptor
289 # TODO feature 1417: Check that no instantiation is set over PDU
290 # check if PDU forces a concrete vim-network-id and add it
291 # check if PDU contains a SDN-assist info (dpid, switch, port) and pass it to RO
293 def vim_account_2_RO(vim_account
):
294 if vim_account
in vim_2_RO
:
295 return vim_2_RO
[vim_account
]
297 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
298 if db_vim
["_admin"]["operationalState"] != "ENABLED":
299 raise LcmException("VIM={} is not available. operationalState={}".format(
300 vim_account
, db_vim
["_admin"]["operationalState"]))
301 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
302 vim_2_RO
[vim_account
] = RO_vim_id
305 def wim_account_2_RO(wim_account
):
306 if isinstance(wim_account
, str):
307 if wim_account
in wim_2_RO
:
308 return wim_2_RO
[wim_account
]
310 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
311 if db_wim
["_admin"]["operationalState"] != "ENABLED":
312 raise LcmException("WIM={} is not available. operationalState={}".format(
313 wim_account
, db_wim
["_admin"]["operationalState"]))
314 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
315 wim_2_RO
[wim_account
] = RO_wim_id
320 def ip_profile_2_RO(ip_profile
):
321 RO_ip_profile
= deepcopy((ip_profile
))
322 if "dns-server" in RO_ip_profile
:
323 if isinstance(RO_ip_profile
["dns-server"], list):
324 RO_ip_profile
["dns-address"] = []
325 for ds
in RO_ip_profile
.pop("dns-server"):
326 RO_ip_profile
["dns-address"].append(ds
['address'])
328 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
329 if RO_ip_profile
.get("ip-version") == "ipv4":
330 RO_ip_profile
["ip-version"] = "IPv4"
331 if RO_ip_profile
.get("ip-version") == "ipv6":
332 RO_ip_profile
["ip-version"] = "IPv6"
333 if "dhcp-params" in RO_ip_profile
:
334 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
340 # "name": ns_params["nsName"],
341 # "description": ns_params.get("nsDescription"),
342 "datacenter": vim_account_2_RO(ns_params
["vimAccountId"]),
343 "wim_account": wim_account_2_RO(ns_params
.get("wimAccountId")),
344 # "scenario": ns_params["nsdId"],
347 n2vc_key_list
= n2vc_key_list
or []
348 for vnfd_ref
, vnfd
in vnfd_dict
.items():
349 vdu_needed_access
= []
351 if vnfd
.get("vnf-configuration"):
352 ssh_required
= deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required"))
353 if ssh_required
and vnfd
.get("mgmt-interface"):
354 if vnfd
["mgmt-interface"].get("vdu-id"):
355 vdu_needed_access
.append(vnfd
["mgmt-interface"]["vdu-id"])
356 elif vnfd
["mgmt-interface"].get("cp"):
357 mgmt_cp
= vnfd
["mgmt-interface"]["cp"]
359 for vdu
in vnfd
.get("vdu", ()):
360 if vdu
.get("vdu-configuration"):
361 ssh_required
= deep_get(vdu
, ("vdu-configuration", "config-access", "ssh-access", "required"))
363 vdu_needed_access
.append(vdu
["id"])
365 for vdu_interface
in vdu
.get("interface"):
366 if vdu_interface
.get("external-connection-point-ref") and \
367 vdu_interface
["external-connection-point-ref"] == mgmt_cp
:
368 vdu_needed_access
.append(vdu
["id"])
372 if vdu_needed_access
:
373 for vnf_member
in nsd
.get("constituent-vnfd"):
374 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
376 for vdu
in vdu_needed_access
:
377 populate_dict(RO_ns_params
,
378 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
, "mgmt_keys"),
381 if ns_params
.get("vduImage"):
382 RO_ns_params
["vduImage"] = ns_params
["vduImage"]
384 if ns_params
.get("ssh_keys"):
385 RO_ns_params
["cloud-config"] = {"key-pairs": ns_params
["ssh_keys"]}
386 for vnf_params
in get_iterable(ns_params
, "vnf"):
387 for constituent_vnfd
in nsd
["constituent-vnfd"]:
388 if constituent_vnfd
["member-vnf-index"] == vnf_params
["member-vnf-index"]:
389 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
392 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index={} is not present at nsd:"
393 "constituent-vnfd".format(vnf_params
["member-vnf-index"]))
394 if vnf_params
.get("vimAccountId"):
395 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "datacenter"),
396 vim_account_2_RO(vnf_params
["vimAccountId"]))
398 for vdu_params
in get_iterable(vnf_params
, "vdu"):
399 # TODO feature 1417: check that this VDU exist and it is not a PDU
400 if vdu_params
.get("volume"):
401 for volume_params
in vdu_params
["volume"]:
402 if volume_params
.get("vim-volume-id"):
403 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
404 vdu_params
["id"], "devices", volume_params
["name"], "vim_id"),
405 volume_params
["vim-volume-id"])
406 if vdu_params
.get("interface"):
407 for interface_params
in vdu_params
["interface"]:
408 if interface_params
.get("ip-address"):
409 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
410 vdu_params
["id"], "interfaces", interface_params
["name"],
412 interface_params
["ip-address"])
413 if interface_params
.get("mac-address"):
414 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
415 vdu_params
["id"], "interfaces", interface_params
["name"],
417 interface_params
["mac-address"])
418 if interface_params
.get("floating-ip-required"):
419 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
420 vdu_params
["id"], "interfaces", interface_params
["name"],
422 interface_params
["floating-ip-required"])
424 for internal_vld_params
in get_iterable(vnf_params
, "internal-vld"):
425 if internal_vld_params
.get("vim-network-name"):
426 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
427 internal_vld_params
["name"], "vim-network-name"),
428 internal_vld_params
["vim-network-name"])
429 if internal_vld_params
.get("vim-network-id"):
430 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
431 internal_vld_params
["name"], "vim-network-id"),
432 internal_vld_params
["vim-network-id"])
433 if internal_vld_params
.get("ip-profile"):
434 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
435 internal_vld_params
["name"], "ip-profile"),
436 ip_profile_2_RO(internal_vld_params
["ip-profile"]))
437 if internal_vld_params
.get("provider-network"):
439 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
440 internal_vld_params
["name"], "provider-network"),
441 internal_vld_params
["provider-network"].copy())
443 for icp_params
in get_iterable(internal_vld_params
, "internal-connection-point"):
446 for vdu_descriptor
in vnf_descriptor
["vdu"]:
447 for vdu_interface
in vdu_descriptor
["interface"]:
448 if vdu_interface
.get("internal-connection-point-ref") == icp_params
["id-ref"]:
449 if icp_params
.get("ip-address"):
450 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
451 vdu_descriptor
["id"], "interfaces",
452 vdu_interface
["name"], "ip_address"),
453 icp_params
["ip-address"])
455 if icp_params
.get("mac-address"):
456 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
457 vdu_descriptor
["id"], "interfaces",
458 vdu_interface
["name"], "mac_address"),
459 icp_params
["mac-address"])
465 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index[{}]:"
466 "internal-vld:id-ref={} is not present at vnfd:internal-"
467 "connection-point".format(vnf_params
["member-vnf-index"],
468 icp_params
["id-ref"]))
470 for vld_params
in get_iterable(ns_params
, "vld"):
471 if "ip-profile" in vld_params
:
472 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "ip-profile"),
473 ip_profile_2_RO(vld_params
["ip-profile"]))
475 if vld_params
.get("provider-network"):
477 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "provider-network"),
478 vld_params
["provider-network"].copy())
480 if "wimAccountId" in vld_params
and vld_params
["wimAccountId"] is not None:
481 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "wim_account"),
482 wim_account_2_RO(vld_params
["wimAccountId"])),
483 if vld_params
.get("vim-network-name"):
485 if isinstance(vld_params
["vim-network-name"], dict):
486 for vim_account
, vim_net
in vld_params
["vim-network-name"].items():
487 RO_vld_sites
.append({
488 "netmap-use": vim_net
,
489 "datacenter": vim_account_2_RO(vim_account
)
491 else: # isinstance str
492 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-name"]})
494 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
496 if vld_params
.get("vim-network-id"):
498 if isinstance(vld_params
["vim-network-id"], dict):
499 for vim_account
, vim_net
in vld_params
["vim-network-id"].items():
500 RO_vld_sites
.append({
501 "netmap-use": vim_net
,
502 "datacenter": vim_account_2_RO(vim_account
)
504 else: # isinstance str
505 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-id"]})
507 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
508 if vld_params
.get("ns-net"):
509 if isinstance(vld_params
["ns-net"], dict):
510 for vld_id
, instance_scenario_id
in vld_params
["ns-net"].items():
511 RO_vld_ns_net
= {"instance_scenario_id": instance_scenario_id
, "osm_id": vld_id
}
512 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "use-network"), RO_vld_ns_net
)
513 if "vnfd-connection-point-ref" in vld_params
:
514 for cp_params
in vld_params
["vnfd-connection-point-ref"]:
516 for constituent_vnfd
in nsd
["constituent-vnfd"]:
517 if constituent_vnfd
["member-vnf-index"] == cp_params
["member-vnf-index-ref"]:
518 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
522 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={} "
523 "is not present at nsd:constituent-vnfd".format(cp_params
["member-vnf-index-ref"]))
525 for vdu_descriptor
in vnf_descriptor
["vdu"]:
526 for interface_descriptor
in vdu_descriptor
["interface"]:
527 if interface_descriptor
.get("external-connection-point-ref") == \
528 cp_params
["vnfd-connection-point-ref"]:
535 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={}:"
536 "vnfd-connection-point-ref={} is not present at vnfd={}".format(
537 cp_params
["member-vnf-index-ref"],
538 cp_params
["vnfd-connection-point-ref"],
539 vnf_descriptor
["id"]))
540 if cp_params
.get("ip-address"):
541 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
542 vdu_descriptor
["id"], "interfaces",
543 interface_descriptor
["name"], "ip_address"),
544 cp_params
["ip-address"])
545 if cp_params
.get("mac-address"):
546 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
547 vdu_descriptor
["id"], "interfaces",
548 interface_descriptor
["name"], "mac_address"),
549 cp_params
["mac-address"])
552 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None):
553 # make a copy to do not change
554 vdu_create
= copy(vdu_create
)
555 vdu_delete
= copy(vdu_delete
)
557 vdurs
= db_vnfr
.get("vdur")
560 vdu_index
= len(vdurs
)
563 vdur
= vdurs
[vdu_index
]
564 if vdur
.get("pdu-type"):
566 vdu_id_ref
= vdur
["vdu-id-ref"]
567 if vdu_create
and vdu_create
.get(vdu_id_ref
):
568 for index
in range(0, vdu_create
[vdu_id_ref
]):
569 vdur
= deepcopy(vdur
)
570 vdur
["_id"] = str(uuid4())
571 vdur
["count-index"] += 1
572 vdurs
.insert(vdu_index
+1+index
, vdur
)
573 del vdu_create
[vdu_id_ref
]
574 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
576 vdu_delete
[vdu_id_ref
] -= 1
577 if not vdu_delete
[vdu_id_ref
]:
578 del vdu_delete
[vdu_id_ref
]
579 # check all operations are done
580 if vdu_create
or vdu_delete
:
581 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
584 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
587 vnfr_update
= {"vdur": vdurs
}
588 db_vnfr
["vdur"] = vdurs
589 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
591 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
593 Updates database nsr with the RO info for the created vld
594 :param ns_update_nsr: dictionary to be filled with the updated info
595 :param db_nsr: content of db_nsr. This is also modified
596 :param nsr_desc_RO: nsr descriptor from RO
597 :return: Nothing, LcmException is raised on errors
600 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
601 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
602 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
604 vld
["vim-id"] = net_RO
.get("vim_net_id")
605 vld
["name"] = net_RO
.get("vim_name")
606 vld
["status"] = net_RO
.get("status")
607 vld
["status-detailed"] = net_RO
.get("error_msg")
608 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
611 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
613 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
615 for db_vnfr
in db_vnfrs
.values():
616 vnfr_update
= {"status": "ERROR"}
617 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
618 if "status" not in vdur
:
619 vdur
["status"] = "ERROR"
620 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
622 vdur
["status-detailed"] = str(error_text
)
623 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
624 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
625 except DbException
as e
:
626 self
.logger
.error("Cannot update vnf. {}".format(e
))
628 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
630 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
631 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
632 :param nsr_desc_RO: nsr descriptor from RO
633 :return: Nothing, LcmException is raised on errors
635 for vnf_index
, db_vnfr
in db_vnfrs
.items():
636 for vnf_RO
in nsr_desc_RO
["vnfs"]:
637 if vnf_RO
["member_vnf_index"] != vnf_index
:
640 if vnf_RO
.get("ip_address"):
641 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
642 elif not db_vnfr
.get("ip-address"):
643 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
644 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
646 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
647 vdur_RO_count_index
= 0
648 if vdur
.get("pdu-type"):
650 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
651 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
653 if vdur
["count-index"] != vdur_RO_count_index
:
654 vdur_RO_count_index
+= 1
656 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
657 if vdur_RO
.get("ip_address"):
658 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
660 vdur
["ip-address"] = None
661 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
662 vdur
["name"] = vdur_RO
.get("vim_name")
663 vdur
["status"] = vdur_RO
.get("status")
664 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
665 for ifacer
in get_iterable(vdur
, "interfaces"):
666 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
667 if ifacer
["name"] == interface_RO
.get("internal_name"):
668 ifacer
["ip-address"] = interface_RO
.get("ip_address")
669 ifacer
["mac-address"] = interface_RO
.get("mac_address")
672 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
674 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
675 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
678 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
679 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
681 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
682 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
683 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
685 vld
["vim-id"] = net_RO
.get("vim_net_id")
686 vld
["name"] = net_RO
.get("vim_name")
687 vld
["status"] = net_RO
.get("status")
688 vld
["status-detailed"] = net_RO
.get("error_msg")
689 vnfr_update
["vld.{}".format(vld_index
)] = vld
692 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
693 vnf_index
, vld
["id"]))
695 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
699 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
701 def _get_ns_config_info(self
, nsr_id
):
703 Generates a mapping between vnf,vdu elements and the N2VC id
704 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
705 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
706 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
707 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
709 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
710 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
712 ns_config_info
= {"osm-config-mapping": mapping
}
713 for vca
in vca_deployed_list
:
714 if not vca
["member-vnf-index"]:
716 if not vca
["vdu_id"]:
717 mapping
[vca
["member-vnf-index"]] = vca
["application"]
719 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
721 return ns_config_info
724 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
):
726 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
727 primitives as verify-ssh-credentials, or config when needed
728 :param desc_primitive_list: information of the descriptor
729 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
730 this element contains a ssh public key
731 :return: The modified list. Can ba an empty list, but always a list
733 if desc_primitive_list
:
734 primitive_list
= desc_primitive_list
.copy()
737 # look for primitive config, and get the position. None if not present
738 config_position
= None
739 for index
, primitive
in enumerate(primitive_list
):
740 if primitive
["name"] == "config":
741 config_position
= index
744 # for NS, add always a config primitive if not present (bug 874)
745 if not vca_deployed
["member-vnf-index"] and config_position
is None:
746 primitive_list
.insert(0, {"name": "config", "parameter": []})
748 # for VNF/VDU add verify-ssh-credentials after config
749 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
750 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
751 return primitive_list
753 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
754 n2vc_key_list
, stage
):
757 RO_descriptor_number
= 0 # number of descriptors created at RO
758 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
759 nslcmop_id
= db_nslcmop
["_id"]
760 start_deploy
= time()
761 ns_params
= db_nslcmop
.get("operationParams")
762 if ns_params
and ns_params
.get("timeout_ns_deploy"):
763 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
765 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
767 # Check for and optionally request placement optimization. Database will be updated if placement activated
768 stage
[2] = "Waiting for Placement."
769 await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
)
773 # get vnfds, instantiate at RO
774 for c_vnf
in nsd
.get("constituent-vnfd", ()):
775 member_vnf_index
= c_vnf
["member-vnf-index"]
776 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
777 vnfd_ref
= vnfd
["id"]
779 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
780 db_nsr_update
["detailed-status"] = " ".join(stage
)
781 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
782 self
._write
_op
_status
(nslcmop_id
, stage
)
784 # self.logger.debug(logging_text + stage[2])
785 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
786 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
787 RO_descriptor_number
+= 1
789 # look position at deployed.RO.vnfd if not present it will be appended at the end
790 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
791 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
794 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
795 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
798 RO_update
= {"member-vnf-index": member_vnf_index
}
799 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
801 RO_update
["id"] = vnfd_list
[0]["uuid"]
802 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
803 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
805 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
806 get("additionalParamsForVnf"), nsr_id
)
807 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
808 RO_update
["id"] = desc
["uuid"]
809 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
810 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
811 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
812 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
817 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
818 db_nsr_update
["detailed-status"] = " ".join(stage
)
819 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
820 self
._write
_op
_status
(nslcmop_id
, stage
)
822 # self.logger.debug(logging_text + stage[2])
823 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
824 RO_descriptor_number
+= 1
825 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
827 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
828 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
829 nsd_ref
, RO_nsd_uuid
))
831 nsd_RO
= deepcopy(nsd
)
832 nsd_RO
["id"] = RO_osm_nsd_id
833 nsd_RO
.pop("_id", None)
834 nsd_RO
.pop("_admin", None)
835 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
836 member_vnf_index
= c_vnf
["member-vnf-index"]
837 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
838 for c_vld
in nsd_RO
.get("vld", ()):
839 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
840 member_vnf_index
= cp
["member-vnf-index-ref"]
841 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
843 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
844 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
845 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
846 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
847 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
850 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
851 db_nsr_update
["detailed-status"] = " ".join(stage
)
852 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
853 self
._write
_op
_status
(nslcmop_id
, stage
)
855 # if present use it unless in error status
856 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
859 stage
[2] = "Looking for existing ns at RO"
860 db_nsr_update
["detailed-status"] = " ".join(stage
)
861 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
862 self
._write
_op
_status
(nslcmop_id
, stage
)
863 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
864 desc
= await self
.RO
.show("ns", RO_nsr_id
)
866 except ROclient
.ROClientException
as e
:
867 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
869 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
871 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
872 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
873 if ns_status
== "ERROR":
874 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
875 self
.logger
.debug(logging_text
+ stage
[2])
876 await self
.RO
.delete("ns", RO_nsr_id
)
877 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
879 stage
[2] = "Checking dependencies"
880 db_nsr_update
["detailed-status"] = " ".join(stage
)
881 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
882 self
._write
_op
_status
(nslcmop_id
, stage
)
883 # self.logger.debug(logging_text + stage[2])
885 # check if VIM is creating and wait look if previous tasks in process
886 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
888 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
889 self
.logger
.debug(logging_text
+ stage
[2])
890 await asyncio
.wait(task_dependency
, timeout
=3600)
891 if ns_params
.get("vnf"):
892 for vnf
in ns_params
["vnf"]:
893 if "vimAccountId" in vnf
:
894 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
897 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
898 self
.logger
.debug(logging_text
+ stage
[2])
899 await asyncio
.wait(task_dependency
, timeout
=3600)
901 stage
[2] = "Checking instantiation parameters."
902 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, n2vc_key_list
)
903 stage
[2] = "Deploying ns at VIM."
904 db_nsr_update
["detailed-status"] = " ".join(stage
)
905 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
906 self
._write
_op
_status
(nslcmop_id
, stage
)
908 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
909 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
910 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
911 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
912 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
914 # wait until NS is ready
915 stage
[2] = "Waiting VIM to deploy ns."
916 db_nsr_update
["detailed-status"] = " ".join(stage
)
917 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
918 self
._write
_op
_status
(nslcmop_id
, stage
)
919 detailed_status_old
= None
920 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
923 while time() <= start_deploy
+ timeout_ns_deploy
:
924 desc
= await self
.RO
.show("ns", RO_nsr_id
)
928 # desc has changed => update db
929 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
932 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
933 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
934 if ns_status
== "ERROR":
935 raise ROclient
.ROClientException(ns_status_info
)
936 elif ns_status
== "BUILD":
937 stage
[2] = "VIM: ({})".format(ns_status_info
)
938 elif ns_status
== "ACTIVE":
939 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
941 self
.ns_update_vnfr(db_vnfrs
, desc
)
943 except LcmExceptionNoMgmtIP
:
946 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
947 if stage
[2] != detailed_status_old
:
948 detailed_status_old
= stage
[2]
949 db_nsr_update
["detailed-status"] = " ".join(stage
)
950 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
951 self
._write
_op
_status
(nslcmop_id
, stage
)
952 await asyncio
.sleep(5, loop
=self
.loop
)
953 else: # timeout_ns_deploy
954 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
957 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
959 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
960 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
961 stage
[2] = "Deployed at VIM"
962 db_nsr_update
["detailed-status"] = " ".join(stage
)
963 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
964 self
._write
_op
_status
(nslcmop_id
, stage
)
965 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
966 # self.logger.debug(logging_text + "Deployed at VIM")
967 except (ROclient
.ROClientException
, LcmException
, DbException
) as e
:
968 stage
[2] = "ERROR deploying at VIM"
969 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
972 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
974 Wait for ip addres at RO, and optionally, insert public key in virtual machine
975 :param logging_text: prefix use for logging
980 :param pub_key: public ssh key to inject, None to skip
981 :param user: user to apply the public ssh key
985 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
995 if ro_retries
>= 360: # 1 hour
996 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
998 await asyncio
.sleep(10, loop
=self
.loop
)
1001 if not target_vdu_id
:
1002 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1004 if not vdu_id
: # for the VNF case
1005 if db_vnfr
.get("status") == "ERROR":
1006 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1007 ip_address
= db_vnfr
.get("ip-address")
1010 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1012 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1013 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1015 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1016 vdur
= db_vnfr
["vdur"][0]
1018 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1021 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1022 ip_address
= vdur
.get("ip-address")
1025 target_vdu_id
= vdur
["vdu-id-ref"]
1026 elif vdur
.get("status") == "ERROR":
1027 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1029 if not target_vdu_id
:
1032 # inject public key into machine
1033 if pub_key
and user
:
1034 # wait until NS is deployed at RO
1036 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1037 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1041 # self.logger.debug(logging_text + "Inserting RO key")
1042 if vdur
.get("pdu-type"):
1043 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1046 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1047 result_dict
= await self
.RO
.create_action(
1049 item_id_name
=ro_nsr_id
,
1050 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1052 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1053 if not result_dict
or not isinstance(result_dict
, dict):
1054 raise LcmException("Unknown response from RO when injecting key")
1055 for result
in result_dict
.values():
1056 if result
.get("vim_result") == 200:
1059 raise ROclient
.ROClientException("error injecting key: {}".format(
1060 result
.get("description")))
1062 except ROclient
.ROClientException
as e
:
1064 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1068 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1074 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1076 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1078 my_vca
= vca_deployed_list
[vca_index
]
1079 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1080 # vdu or kdu: no dependencies
1084 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1085 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1086 configuration_status_list
= db_nsr
["configurationStatus"]
1087 for index
, vca_deployed
in enumerate(configuration_status_list
):
1088 if index
== vca_index
:
1091 if not my_vca
.get("member-vnf-index") or \
1092 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1093 internal_status
= configuration_status_list
[index
].get("status")
1094 if internal_status
== 'READY':
1096 elif internal_status
== 'BROKEN':
1097 raise LcmException("Configuration aborted because dependent charm/s has failed")
1101 # no dependencies, return
1103 await asyncio
.sleep(10)
1106 raise LcmException("Configuration aborted because dependent charm/s timeout")
1108 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1109 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
):
1110 nsr_id
= db_nsr
["_id"]
1111 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1112 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1113 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1115 'collection': 'nsrs',
1116 'filter': {'_id': nsr_id
},
1117 'path': db_update_entry
1123 element_under_configuration
= nsr_id
1127 vnfr_id
= db_vnfr
["_id"]
1129 namespace
= "{nsi}.{ns}".format(
1130 nsi
=nsi_id
if nsi_id
else "",
1134 element_type
= 'VNF'
1135 element_under_configuration
= vnfr_id
1136 namespace
+= ".{}".format(vnfr_id
)
1138 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1139 element_type
= 'VDU'
1140 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1142 namespace
+= ".{}".format(kdu_name
)
1143 element_type
= 'KDU'
1144 element_under_configuration
= kdu_name
1147 self
.fs
.sync() # Sync from FSMongo
1148 artifact_path
= "{}/{}/charms/{}".format(
1149 base_folder
["folder"],
1150 base_folder
["pkg-dir"],
1151 config_descriptor
["juju"]["charm"]
1154 is_proxy_charm
= deep_get(config_descriptor
, ('juju', 'charm')) is not None
1155 if deep_get(config_descriptor
, ('juju', 'proxy')) is False:
1156 is_proxy_charm
= False
1158 # n2vc_redesign STEP 3.1
1160 # find old ee_id if exists
1161 ee_id
= vca_deployed
.get("ee_id")
1163 # create or register execution environment in VCA
1166 self
._write
_configuration
_status
(
1168 vca_index
=vca_index
,
1170 element_under_configuration
=element_under_configuration
,
1171 element_type
=element_type
1174 step
= "create execution environment"
1175 self
.logger
.debug(logging_text
+ step
)
1176 ee_id
, credentials
= await self
.n2vc
.create_execution_environment(namespace
=namespace
,
1181 step
= "Waiting to VM being up and getting IP address"
1182 self
.logger
.debug(logging_text
+ step
)
1183 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1184 user
=None, pub_key
=None)
1185 credentials
= {"hostname": rw_mgmt_ip
}
1187 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1188 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1189 # merged. Meanwhile let's get username from initial-config-primitive
1190 if not username
and config_descriptor
.get("initial-config-primitive"):
1191 for config_primitive
in config_descriptor
["initial-config-primitive"]:
1192 for param
in config_primitive
.get("parameter", ()):
1193 if param
["name"] == "ssh-username":
1194 username
= param
["value"]
1197 raise LcmException("Cannot determine the username neither with 'initial-config-promitive' nor with "
1198 "'config-access.ssh-access.default-user'")
1199 credentials
["username"] = username
1200 # n2vc_redesign STEP 3.2
1202 self
._write
_configuration
_status
(
1204 vca_index
=vca_index
,
1205 status
='REGISTERING',
1206 element_under_configuration
=element_under_configuration
,
1207 element_type
=element_type
1210 step
= "register execution environment {}".format(credentials
)
1211 self
.logger
.debug(logging_text
+ step
)
1212 ee_id
= await self
.n2vc
.register_execution_environment(credentials
=credentials
, namespace
=namespace
,
1215 # for compatibility with MON/POL modules, the need model and application name at database
1216 # TODO ask to N2VC instead of assuming the format "model_name.application_name"
1217 ee_id_parts
= ee_id
.split('.')
1218 model_name
= ee_id_parts
[0]
1219 application_name
= ee_id_parts
[1]
1220 db_nsr_update
= {db_update_entry
+ "model": model_name
,
1221 db_update_entry
+ "application": application_name
,
1222 db_update_entry
+ "ee_id": ee_id
}
1224 # n2vc_redesign STEP 3.3
1226 step
= "Install configuration Software"
1228 self
._write
_configuration
_status
(
1230 vca_index
=vca_index
,
1231 status
='INSTALLING SW',
1232 element_under_configuration
=element_under_configuration
,
1233 element_type
=element_type
,
1234 other_update
=db_nsr_update
1237 # TODO check if already done
1238 self
.logger
.debug(logging_text
+ step
)
1240 if not is_proxy_charm
:
1241 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1242 if initial_config_primitive_list
:
1243 for primitive
in initial_config_primitive_list
:
1244 if primitive
["name"] == "config":
1245 config
= self
._map
_primitive
_params
(
1251 await self
.n2vc
.install_configuration_sw(
1253 artifact_path
=artifact_path
,
1258 # write in db flag of configuration_sw already installed
1259 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1261 # add relations for this VCA (wait for other peers related with this VCA)
1262 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
, vca_index
=vca_index
)
1264 # if SSH access is required, then get execution environment SSH public
1265 if is_proxy_charm
: # if native charm we have waited already to VM be UP
1268 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1269 # Needed to inject a ssh key
1270 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1271 step
= "Install configuration Software, getting public ssh key"
1272 pub_key
= await self
.n2vc
.get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1274 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1276 step
= "Waiting to VM being up and getting IP address"
1277 self
.logger
.debug(logging_text
+ step
)
1279 # n2vc_redesign STEP 5.1
1280 # wait for RO (ip-address) Insert pub_key into VM
1282 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1283 user
=user
, pub_key
=pub_key
)
1285 rw_mgmt_ip
= None # This is for a NS configuration
1287 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1289 # store rw_mgmt_ip in deploy params for later replacement
1290 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1292 # n2vc_redesign STEP 6 Execute initial config primitive
1293 step
= 'execute initial config primitive'
1294 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1296 # sort initial config primitives by 'seq'
1297 if initial_config_primitive_list
:
1299 initial_config_primitive_list
.sort(key
=lambda val
: int(val
['seq']))
1300 except Exception as e
:
1301 self
.logger
.error(logging_text
+ step
+ ": " + str(e
))
1303 self
.logger
.debug(logging_text
+ step
+ ": No initial-config-primitive")
1305 # add config if not present for NS charm
1306 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1309 # wait for dependent primitives execution (NS -> VNF -> VDU)
1310 if initial_config_primitive_list
:
1311 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1313 # stage, in function of element type: vdu, kdu, vnf or ns
1314 my_vca
= vca_deployed_list
[vca_index
]
1315 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1317 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1318 elif my_vca
.get("member-vnf-index"):
1320 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1323 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1325 self
._write
_configuration
_status
(
1327 vca_index
=vca_index
,
1328 status
='EXECUTING PRIMITIVE'
1331 self
._write
_op
_status
(
1336 check_if_terminated_needed
= True
1337 for initial_config_primitive
in initial_config_primitive_list
:
1338 # adding information on the vca_deployed if it is a NS execution environment
1339 if not vca_deployed
["member-vnf-index"]:
1340 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1341 # TODO check if already done
1342 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1344 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1345 self
.logger
.debug(logging_text
+ step
)
1346 await self
.n2vc
.exec_primitive(
1348 primitive_name
=initial_config_primitive
["name"],
1349 params_dict
=primitive_params_
,
1352 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1353 if check_if_terminated_needed
:
1354 if config_descriptor
.get('terminate-config-primitive'):
1355 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1356 check_if_terminated_needed
= False
1358 # TODO register in database that primitive is done
1360 step
= "instantiated at VCA"
1361 self
.logger
.debug(logging_text
+ step
)
1363 self
._write
_configuration
_status
(
1365 vca_index
=vca_index
,
1369 except Exception as e
: # TODO not use Exception but N2VC exception
1370 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1371 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1372 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1373 self
._write
_configuration
_status
(
1375 vca_index
=vca_index
,
1378 raise LcmException("{} {}".format(step
, e
)) from e
1380 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1381 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1383 Update db_nsr fields.
1386 :param current_operation:
1387 :param current_operation_id:
1388 :param error_description:
1389 :param error_detail:
1390 :param other_update: Other required changes at database if provided, will be cleared
1394 db_dict
= other_update
or {}
1395 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1396 db_dict
["_admin.current-operation"] = current_operation_id
1397 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1398 db_dict
["currentOperation"] = current_operation
1399 db_dict
["currentOperationID"] = current_operation_id
1400 db_dict
["errorDescription"] = error_description
1401 db_dict
["errorDetail"] = error_detail
1404 db_dict
["nsState"] = ns_state
1405 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1406 except DbException
as e
:
1407 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1409 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1410 operation_state
: str = None, other_update
: dict = None):
1412 db_dict
= other_update
or {}
1413 db_dict
['queuePosition'] = queuePosition
1414 if isinstance(stage
, list):
1415 db_dict
['stage'] = stage
[0]
1416 db_dict
['detailed-status'] = " ".join(stage
)
1417 elif stage
is not None:
1418 db_dict
['stage'] = str(stage
)
1420 if error_message
is not None:
1421 db_dict
['errorMessage'] = error_message
1422 if operation_state
is not None:
1423 db_dict
['operationState'] = operation_state
1424 db_dict
["statusEnteredTime"] = time()
1425 self
.update_db_2("nslcmops", op_id
, db_dict
)
1426 except DbException
as e
:
1427 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1429 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1431 nsr_id
= db_nsr
["_id"]
1432 # configurationStatus
1433 config_status
= db_nsr
.get('configurationStatus')
1435 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1436 enumerate(config_status
) if v
}
1438 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1440 except DbException
as e
:
1441 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1443 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1444 element_under_configuration
: str = None, element_type
: str = None,
1445 other_update
: dict = None):
1447 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1448 # .format(vca_index, status))
1451 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1452 db_dict
= other_update
or {}
1454 db_dict
[db_path
+ 'status'] = status
1455 if element_under_configuration
:
1456 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1458 db_dict
[db_path
+ 'elementType'] = element_type
1459 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1460 except DbException
as e
:
1461 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1462 .format(status
, nsr_id
, vca_index
, e
))
1464 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1466 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1467 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1468 Database is used because the result can be obtained from a different LCM worker in case of HA.
1469 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1470 :param db_nslcmop: database content of nslcmop
1471 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1472 :return: None. Modifies database vnfrs and parameter db_vnfr with the computed 'vim-account-id'
1474 nslcmop_id
= db_nslcmop
['_id']
1475 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1476 if placement_engine
== "PLA":
1477 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1478 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1479 db_poll_interval
= 5
1480 wait
= db_poll_interval
* 10
1482 while not pla_result
and wait
>= 0:
1483 await asyncio
.sleep(db_poll_interval
)
1484 wait
-= db_poll_interval
1485 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1486 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1489 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1491 for pla_vnf
in pla_result
['vnf']:
1492 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1493 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1495 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1497 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1500 def update_nsrs_with_pla_result(self
, params
):
1502 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1503 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1504 except Exception as e
:
1505 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1507 async def instantiate(self
, nsr_id
, nslcmop_id
):
1510 :param nsr_id: ns instance to deploy
1511 :param nslcmop_id: operation to run
1515 # Try to lock HA task here
1516 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1517 if not task_is_locked_by_me
:
1518 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1521 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1522 self
.logger
.debug(logging_text
+ "Enter")
1524 # get all needed from database
1526 # database nsrs record
1529 # database nslcmops record
1532 # update operation on nsrs
1534 # update operation on nslcmops
1535 db_nslcmop_update
= {}
1537 nslcmop_operation_state
= None
1538 db_vnfrs
= {} # vnf's info indexed by member-index
1540 tasks_dict_info
= {} # from task to info text
1543 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1544 # ^ stage, step, VIM progress
1546 # wait for any previous tasks in process
1547 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1549 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1550 stage
[1] = "Reading from database,"
1551 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1552 db_nsr_update
["detailed-status"] = "creating"
1553 db_nsr_update
["operational-status"] = "init"
1554 self
._write
_ns
_status
(
1556 ns_state
="BUILDING",
1557 current_operation
="INSTANTIATING",
1558 current_operation_id
=nslcmop_id
,
1559 other_update
=db_nsr_update
1561 self
._write
_op
_status
(
1567 # read from db: operation
1568 stage
[1] = "Getting nslcmop={} from db".format(nslcmop_id
)
1569 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1570 ns_params
= db_nslcmop
.get("operationParams")
1571 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1572 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1574 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1577 stage
[1] = "Getting nsr={} from db".format(nsr_id
)
1578 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1579 # nsd is replicated into ns (no db read)
1581 # nsr_name = db_nsr["name"] # TODO short-name??
1583 # read from db: vnf's of this ns
1584 stage
[1] = "Getting vnfrs from db"
1585 self
.logger
.debug(logging_text
+ stage
[1])
1586 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1588 # read from db: vnfd's for every vnf
1589 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1590 db_vnfds
= {} # every vnfd data indexed by vnf id
1591 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1593 # for each vnf in ns, read vnfd
1594 for vnfr
in db_vnfrs_list
:
1595 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1596 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1597 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1598 # if we haven't this vnfd, read it from db
1599 if vnfd_id
not in db_vnfds
:
1601 stage
[1] = "Getting vnfd={} id='{}' from db".format(vnfd_id
, vnfd_ref
)
1602 self
.logger
.debug(logging_text
+ stage
[1])
1603 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1606 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1607 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1608 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1610 # Get or generates the _admin.deployed.VCA list
1611 vca_deployed_list
= None
1612 if db_nsr
["_admin"].get("deployed"):
1613 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1614 if vca_deployed_list
is None:
1615 vca_deployed_list
= []
1616 configuration_status_list
= []
1617 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1618 db_nsr_update
["configurationStatus"] = configuration_status_list
1619 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1620 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1621 elif isinstance(vca_deployed_list
, dict):
1622 # maintain backward compatibility. Change a dict to list at database
1623 vca_deployed_list
= list(vca_deployed_list
.values())
1624 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1625 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1627 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1628 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1629 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1631 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1632 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1633 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1635 # n2vc_redesign STEP 2 Deploy Network Scenario
1636 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1637 self
._write
_op
_status
(
1642 stage
[1] = "Deploying KDUs,"
1643 # self.logger.debug(logging_text + "Before deploy_kdus")
1644 # Call to deploy_kdus in case exists the "vdu:kdu" param
1645 await self
.deploy_kdus(
1646 logging_text
=logging_text
,
1648 nslcmop_id
=nslcmop_id
,
1651 task_instantiation_info
=tasks_dict_info
,
1654 stage
[1] = "Getting VCA public key."
1655 # n2vc_redesign STEP 1 Get VCA public ssh-key
1656 # feature 1429. Add n2vc public key to needed VMs
1657 n2vc_key
= self
.n2vc
.get_public_key()
1658 n2vc_key_list
= [n2vc_key
]
1659 if self
.vca_config
.get("public_key"):
1660 n2vc_key_list
.append(self
.vca_config
["public_key"])
1662 stage
[1] = "Deploying NS at VIM."
1663 task_ro
= asyncio
.ensure_future(
1664 self
.instantiate_RO(
1665 logging_text
=logging_text
,
1669 db_nslcmop
=db_nslcmop
,
1671 db_vnfds_ref
=db_vnfds_ref
,
1672 n2vc_key_list
=n2vc_key_list
,
1676 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
1677 tasks_dict_info
[task_ro
] = "Deploying at VIM"
1679 # n2vc_redesign STEP 3 to 6 Deploy N2VC
1680 stage
[1] = "Deploying Execution Environments."
1681 self
.logger
.debug(logging_text
+ stage
[1])
1683 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
1684 # get_iterable() returns a value from a dict or empty tuple if key does not exist
1685 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
1686 vnfd_id
= c_vnf
["vnfd-id-ref"]
1687 vnfd
= db_vnfds_ref
[vnfd_id
]
1688 member_vnf_index
= str(c_vnf
["member-vnf-index"])
1689 db_vnfr
= db_vnfrs
[member_vnf_index
]
1690 base_folder
= vnfd
["_admin"]["storage"]
1696 # Get additional parameters
1698 if db_vnfr
.get("additionalParamsForVnf"):
1699 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
1701 descriptor_config
= vnfd
.get("vnf-configuration")
1702 if descriptor_config
and descriptor_config
.get("juju"):
1704 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
1707 nslcmop_id
=nslcmop_id
,
1713 member_vnf_index
=member_vnf_index
,
1714 vdu_index
=vdu_index
,
1716 deploy_params
=deploy_params
,
1717 descriptor_config
=descriptor_config
,
1718 base_folder
=base_folder
,
1719 task_instantiation_info
=tasks_dict_info
,
1723 # Deploy charms for each VDU that supports one.
1724 for vdud
in get_iterable(vnfd
, 'vdu'):
1726 descriptor_config
= vdud
.get('vdu-configuration')
1727 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
1728 if vdur
.get("additionalParams"):
1729 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
1731 deploy_params_vdu
= deploy_params
1732 if descriptor_config
and descriptor_config
.get("juju"):
1733 # look for vdu index in the db_vnfr["vdu"] section
1734 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1735 # if vdur["vdu-id-ref"] == vdu_id:
1738 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1739 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1740 # vdu_name = vdur.get("name")
1743 for vdu_index
in range(int(vdud
.get("count", 1))):
1744 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
1746 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
1747 member_vnf_index
, vdu_id
, vdu_index
),
1750 nslcmop_id
=nslcmop_id
,
1756 member_vnf_index
=member_vnf_index
,
1757 vdu_index
=vdu_index
,
1759 deploy_params
=deploy_params_vdu
,
1760 descriptor_config
=descriptor_config
,
1761 base_folder
=base_folder
,
1762 task_instantiation_info
=tasks_dict_info
,
1765 for kdud
in get_iterable(vnfd
, 'kdu'):
1766 kdu_name
= kdud
["name"]
1767 descriptor_config
= kdud
.get('kdu-configuration')
1768 if descriptor_config
and descriptor_config
.get("juju"):
1772 # look for vdu index in the db_vnfr["vdu"] section
1773 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1774 # if vdur["vdu-id-ref"] == vdu_id:
1777 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1778 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1779 # vdu_name = vdur.get("name")
1783 logging_text
=logging_text
,
1786 nslcmop_id
=nslcmop_id
,
1792 member_vnf_index
=member_vnf_index
,
1793 vdu_index
=vdu_index
,
1795 deploy_params
=deploy_params
,
1796 descriptor_config
=descriptor_config
,
1797 base_folder
=base_folder
,
1798 task_instantiation_info
=tasks_dict_info
,
1802 # Check if this NS has a charm configuration
1803 descriptor_config
= nsd
.get("ns-configuration")
1804 if descriptor_config
and descriptor_config
.get("juju"):
1807 member_vnf_index
= None
1813 # Get additional parameters
1815 if db_nsr
.get("additionalParamsForNs"):
1816 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
1817 base_folder
= nsd
["_admin"]["storage"]
1819 logging_text
=logging_text
,
1822 nslcmop_id
=nslcmop_id
,
1828 member_vnf_index
=member_vnf_index
,
1829 vdu_index
=vdu_index
,
1831 deploy_params
=deploy_params
,
1832 descriptor_config
=descriptor_config
,
1833 base_folder
=base_folder
,
1834 task_instantiation_info
=tasks_dict_info
,
1838 # rest of staff will be done at finally
1840 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
1841 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
1843 except asyncio
.CancelledError
:
1844 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
1845 exc
= "Operation was cancelled"
1846 except Exception as e
:
1847 exc
= traceback
.format_exc()
1848 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
1851 error_list
.append(str(exc
))
1853 # wait for pending tasks
1855 stage
[1] = "Waiting for instantiate pending tasks."
1856 self
.logger
.debug(logging_text
+ stage
[1])
1857 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
1858 stage
, nslcmop_id
, nsr_id
=nsr_id
)
1859 stage
[1] = stage
[2] = ""
1860 except asyncio
.CancelledError
:
1861 error_list
.append("Cancelled")
1862 # TODO cancel all tasks
1863 except Exception as exc
:
1864 error_list
.append(str(exc
))
1866 # update operation-status
1867 db_nsr_update
["operational-status"] = "running"
1868 # let's begin with VCA 'configured' status (later we can change it)
1869 db_nsr_update
["config-status"] = "configured"
1870 for task
, task_name
in tasks_dict_info
.items():
1871 if not task
.done() or task
.cancelled() or task
.exception():
1872 if task_name
.startswith(self
.task_name_deploy_vca
):
1873 # A N2VC task is pending
1874 db_nsr_update
["config-status"] = "failed"
1876 # RO or KDU task is pending
1877 db_nsr_update
["operational-status"] = "failed"
1879 # update status at database
1881 error_detail
= ". ".join(error_list
)
1882 self
.logger
.error(logging_text
+ error_detail
)
1883 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
1884 error_description_nsr
= 'Operation: INSTANTIATING.{}, Stage {}'.format(nslcmop_id
, stage
[0])
1886 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
1887 db_nslcmop_update
["detailed-status"] = error_detail
1888 nslcmop_operation_state
= "FAILED"
1892 error_description_nsr
= error_description_nslcmop
= None
1894 db_nsr_update
["detailed-status"] = "Done"
1895 db_nslcmop_update
["detailed-status"] = "Done"
1896 nslcmop_operation_state
= "COMPLETED"
1899 self
._write
_ns
_status
(
1902 current_operation
="IDLE",
1903 current_operation_id
=None,
1904 error_description
=error_description_nsr
,
1905 error_detail
=error_detail
,
1906 other_update
=db_nsr_update
1909 self
._write
_op
_status
(
1912 error_message
=error_description_nslcmop
,
1913 operation_state
=nslcmop_operation_state
,
1914 other_update
=db_nslcmop_update
,
1917 if nslcmop_operation_state
:
1919 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
1920 "operationState": nslcmop_operation_state
},
1922 except Exception as e
:
1923 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
1925 self
.logger
.debug(logging_text
+ "Exit")
1926 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
1928 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int, timeout
: int = 3600) -> bool:
1931 # 1. find all relations for this VCA
1932 # 2. wait for other peers related
1937 # STEP 1: find all relations for this VCA
1940 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1943 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
1945 # read all ns-configuration relations
1946 ns_relations
= list()
1947 db_ns_relations
= deep_get(db_nsr
, ('nsd', 'ns-configuration', 'relation'))
1949 for r
in db_ns_relations
:
1950 # check if this VCA is in the relation
1951 if my_vca
.get('member-vnf-index') in\
1952 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1953 ns_relations
.append(r
)
1955 # read all vnf-configuration relations
1956 vnf_relations
= list()
1957 db_vnfd_list
= db_nsr
.get('vnfd-id')
1959 for vnfd
in db_vnfd_list
:
1960 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
1961 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
1962 if db_vnf_relations
:
1963 for r
in db_vnf_relations
:
1964 # check if this VCA is in the relation
1965 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1966 vnf_relations
.append(r
)
1968 # if no relations, terminate
1969 if not ns_relations
and not vnf_relations
:
1970 self
.logger
.debug(logging_text
+ ' No relations')
1973 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
1980 if now
- start
>= timeout
:
1981 self
.logger
.error(logging_text
+ ' : timeout adding relations')
1984 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
1985 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1987 # for each defined NS relation, find the VCA's related
1988 for r
in ns_relations
:
1989 from_vca_ee_id
= None
1991 from_vca_endpoint
= None
1992 to_vca_endpoint
= None
1993 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
1994 for vca
in vca_list
:
1995 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
1996 and vca
.get('config_sw_installed'):
1997 from_vca_ee_id
= vca
.get('ee_id')
1998 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
1999 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2000 and vca
.get('config_sw_installed'):
2001 to_vca_ee_id
= vca
.get('ee_id')
2002 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2003 if from_vca_ee_id
and to_vca_ee_id
:
2005 await self
.n2vc
.add_relation(
2006 ee_id_1
=from_vca_ee_id
,
2007 ee_id_2
=to_vca_ee_id
,
2008 endpoint_1
=from_vca_endpoint
,
2009 endpoint_2
=to_vca_endpoint
)
2010 # remove entry from relations list
2011 ns_relations
.remove(r
)
2013 # check failed peers
2015 vca_status_list
= db_nsr
.get('configurationStatus')
2017 for i
in range(len(vca_list
)):
2019 vca_status
= vca_status_list
[i
]
2020 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2021 if vca_status
.get('status') == 'BROKEN':
2022 # peer broken: remove relation from list
2023 ns_relations
.remove(r
)
2024 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2025 if vca_status
.get('status') == 'BROKEN':
2026 # peer broken: remove relation from list
2027 ns_relations
.remove(r
)
2032 # for each defined VNF relation, find the VCA's related
2033 for r
in vnf_relations
:
2034 from_vca_ee_id
= None
2036 from_vca_endpoint
= None
2037 to_vca_endpoint
= None
2038 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2039 for vca
in vca_list
:
2040 if vca
.get('vdu_id') == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2041 from_vca_ee_id
= vca
.get('ee_id')
2042 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2043 if vca
.get('vdu_id') == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2044 to_vca_ee_id
= vca
.get('ee_id')
2045 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2046 if from_vca_ee_id
and to_vca_ee_id
:
2048 await self
.n2vc
.add_relation(
2049 ee_id_1
=from_vca_ee_id
,
2050 ee_id_2
=to_vca_ee_id
,
2051 endpoint_1
=from_vca_endpoint
,
2052 endpoint_2
=to_vca_endpoint
)
2053 # remove entry from relations list
2054 vnf_relations
.remove(r
)
2056 # check failed peers
2058 vca_status_list
= db_nsr
.get('configurationStatus')
2060 for i
in range(len(vca_list
)):
2062 vca_status
= vca_status_list
[i
]
2063 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2064 if vca_status
.get('status') == 'BROKEN':
2065 # peer broken: remove relation from list
2066 ns_relations
.remove(r
)
2067 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2068 if vca_status
.get('status') == 'BROKEN':
2069 # peer broken: remove relation from list
2070 ns_relations
.remove(r
)
2076 await asyncio
.sleep(5.0)
2078 if not ns_relations
and not vnf_relations
:
2079 self
.logger
.debug('Relations added')
2084 except Exception as e
:
2085 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2088 def _write_db_callback(self
, task
, item
, _id
, on_done
=None, on_exc
=None):
2090 callback for kdu install intended to store the returned kdu_instance at database
2095 result
= task
.result()
2097 db_update
[on_done
] = str(result
)
2098 except Exception as e
:
2100 db_update
[on_exc
] = str(e
)
2103 self
.update_db_2(item
, _id
, db_update
)
2107 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2108 # Launch kdus if present in the descriptor
2110 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2112 def _get_cluster_id(cluster_id
, cluster_type
):
2113 nonlocal k8scluster_id_2_uuic
2114 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2115 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2117 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2118 if not db_k8scluster
:
2119 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2120 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2122 raise LcmException("K8s cluster '{}' has not been initilized for '{}'".format(cluster_id
, cluster_type
))
2123 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2126 logging_text
+= "Deploy kdus: "
2129 db_nsr_update
= {"_admin.deployed.K8s": []}
2130 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2133 updated_cluster_list
= []
2135 for vnfr_data
in db_vnfrs
.values():
2136 for kdur
in get_iterable(vnfr_data
, "kdur"):
2137 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2138 vnfd_id
= vnfr_data
.get('vnfd-id')
2139 namespace
= kdur
.get("k8s-namespace")
2140 if kdur
.get("helm-chart"):
2141 kdumodel
= kdur
["helm-chart"]
2142 k8sclustertype
= "helm-chart"
2143 elif kdur
.get("juju-bundle"):
2144 kdumodel
= kdur
["juju-bundle"]
2145 k8sclustertype
= "juju-bundle"
2147 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2148 "juju-bundle. Maybe an old NBI version is running".
2149 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2150 # check if kdumodel is a file and exists
2152 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2153 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2154 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2155 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["'pkg-dir"], k8sclustertype
,
2157 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2158 kdumodel
= self
.fs
.path
+ filename
2159 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2161 except Exception: # it is not a file
2164 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2165 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2166 cluster_uuid
= _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2168 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2169 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2170 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2171 if del_repo_list
or added_repo_dict
:
2172 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2173 updated
= {'_admin.helm_charts_added.' +
2174 item
: name
for item
, name
in added_repo_dict
.items()}
2175 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2176 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2178 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2179 updated_cluster_list
.append(cluster_uuid
)
2181 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2182 kdur
["kdu-name"], k8s_cluster_id
)
2184 k8s_instace_info
= {"kdu-instance": None,
2185 "k8scluster-uuid": cluster_uuid
,
2186 "k8scluster-type": k8sclustertype
,
2187 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2188 "kdu-name": kdur
["kdu-name"],
2189 "kdu-model": kdumodel
,
2190 "namespace": namespace
}
2191 db_path
= "_admin.deployed.K8s.{}".format(index
)
2192 db_nsr_update
[db_path
] = k8s_instace_info
2193 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2195 db_dict
= {"collection": "nsrs",
2196 "filter": {"_id": nsr_id
},
2199 task
= asyncio
.ensure_future(
2200 self
.k8scluster_map
[k8sclustertype
].install(cluster_uuid
=cluster_uuid
, kdu_model
=kdumodel
,
2201 atomic
=True, params
=desc_params
,
2202 db_dict
=db_dict
, timeout
=600,
2203 kdu_name
=kdur
["kdu-name"], namespace
=namespace
))
2205 task
.add_done_callback(partial(self
._write
_db
_callback
, item
="nsrs", _id
=nsr_id
,
2206 on_done
=db_path
+ ".kdu-instance",
2207 on_exc
=db_path
+ ".detailed-status"))
2208 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2209 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2213 except (LcmException
, asyncio
.CancelledError
):
2215 except Exception as e
:
2216 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2217 if isinstance(e
, (N2VCException
, DbException
)):
2218 self
.logger
.error(logging_text
+ msg
)
2220 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2221 raise LcmException(msg
)
2224 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2226 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2227 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2228 base_folder
, task_instantiation_info
, stage
):
2229 # launch instantiate_N2VC in a asyncio task and register task object
2230 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2231 # if not found, create one entry and update database
2233 # fill db_nsr._admin.deployed.VCA.<index>
2235 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2236 if not vca_deployed
:
2238 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2239 vca_deployed
.get("vdu_id") == vdu_id
and \
2240 vca_deployed
.get("kdu_name") == kdu_name
and \
2241 vca_deployed
.get("vdu_count_index", 0) == vdu_index
:
2244 # not found, create one.
2246 "member-vnf-index": member_vnf_index
,
2248 "kdu_name": kdu_name
,
2249 "vdu_count_index": vdu_index
,
2250 "operational-status": "init", # TODO revise
2251 "detailed-status": "", # TODO revise
2252 "step": "initial-deploy", # TODO revise
2254 "vdu_name": vdu_name
,
2258 # create VCA and configurationStatus in db
2260 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2261 "configurationStatus.{}".format(vca_index
): dict()
2263 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2265 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2268 task_n2vc
= asyncio
.ensure_future(
2269 self
.instantiate_N2VC(
2270 logging_text
=logging_text
,
2271 vca_index
=vca_index
,
2277 vdu_index
=vdu_index
,
2278 deploy_params
=deploy_params
,
2279 config_descriptor
=descriptor_config
,
2280 base_folder
=base_folder
,
2281 nslcmop_id
=nslcmop_id
,
2285 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2286 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2287 member_vnf_index
or "", vdu_id
or "")
2289 # Check if this VNFD has a configured terminate action
2290 def _has_terminate_config_primitive(self
, vnfd
):
2291 vnf_config
= vnfd
.get("vnf-configuration")
2292 if vnf_config
and vnf_config
.get("terminate-config-primitive"):
2298 def _get_terminate_config_primitive_seq_list(vnfd
):
2299 """ Get a numerically sorted list of the sequences for this VNFD's terminate action """
2300 # No need to check for existing primitive twice, already done before
2301 vnf_config
= vnfd
.get("vnf-configuration")
2302 seq_list
= vnf_config
.get("terminate-config-primitive")
2303 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2304 seq_list_sorted
= sorted(seq_list
, key
=lambda x
: int(x
['seq']))
2305 return seq_list_sorted
2308 def _create_nslcmop(nsr_id
, operation
, params
):
2310 Creates a ns-lcm-opp content to be stored at database.
2311 :param nsr_id: internal id of the instance
2312 :param operation: instantiate, terminate, scale, action, ...
2313 :param params: user parameters for the operation
2314 :return: dictionary following SOL005 format
2316 # Raise exception if invalid arguments
2317 if not (nsr_id
and operation
and params
):
2319 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2325 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2326 "operationState": "PROCESSING",
2327 "statusEnteredTime": now
,
2328 "nsInstanceId": nsr_id
,
2329 "lcmOperationType": operation
,
2331 "isAutomaticInvocation": False,
2332 "operationParams": params
,
2333 "isCancelPending": False,
2335 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2336 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2341 def _format_additional_params(self
, params
):
2342 params
= params
or {}
2343 for key
, value
in params
.items():
2344 if str(value
).startswith("!!yaml "):
2345 params
[key
] = yaml
.safe_load(value
[7:])
2348 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2349 primitive
= seq
.get('name')
2350 primitive_params
= {}
2352 "member_vnf_index": vnf_index
,
2353 "primitive": primitive
,
2354 "primitive_params": primitive_params
,
2357 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2361 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2362 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2363 if op
.get('operationState') == 'COMPLETED':
2364 # b. Skip sub-operation
2365 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2366 return self
.SUBOPERATION_STATUS_SKIP
2368 # c. Reintent executing sub-operation
2369 # The sub-operation exists, and operationState != 'COMPLETED'
2370 # Update operationState = 'PROCESSING' to indicate a reintent.
2371 operationState
= 'PROCESSING'
2372 detailed_status
= 'In progress'
2373 self
._update
_suboperation
_status
(
2374 db_nslcmop
, op_index
, operationState
, detailed_status
)
2375 # Return the sub-operation index
2376 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2377 # with arguments extracted from the sub-operation
2380 # Find a sub-operation where all keys in a matching dictionary must match
2381 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2382 def _find_suboperation(self
, db_nslcmop
, match
):
2383 if (db_nslcmop
and match
):
2384 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2385 for i
, op
in enumerate(op_list
):
2386 if all(op
.get(k
) == match
[k
] for k
in match
):
2388 return self
.SUBOPERATION_STATUS_NOT_FOUND
2390 # Update status for a sub-operation given its index
2391 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2392 # Update DB for HA tasks
2393 q_filter
= {'_id': db_nslcmop
['_id']}
2394 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2395 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2396 self
.db
.set_one("nslcmops",
2398 update_dict
=update_dict
,
2399 fail_on_empty
=False)
2401 # Add sub-operation, return the index of the added sub-operation
2402 # Optionally, set operationState, detailed-status, and operationType
2403 # Status and type are currently set for 'scale' sub-operations:
2404 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2405 # 'detailed-status' : status message
2406 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2407 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2408 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2409 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2410 RO_nsr_id
=None, RO_scaling_info
=None):
2412 return self
.SUBOPERATION_STATUS_NOT_FOUND
2413 # Get the "_admin.operations" list, if it exists
2414 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2415 op_list
= db_nslcmop_admin
.get('operations')
2416 # Create or append to the "_admin.operations" list
2417 new_op
= {'member_vnf_index': vnf_index
,
2419 'vdu_count_index': vdu_count_index
,
2420 'primitive': primitive
,
2421 'primitive_params': mapped_primitive_params
}
2423 new_op
['operationState'] = operationState
2425 new_op
['detailed-status'] = detailed_status
2427 new_op
['lcmOperationType'] = operationType
2429 new_op
['RO_nsr_id'] = RO_nsr_id
2431 new_op
['RO_scaling_info'] = RO_scaling_info
2433 # No existing operations, create key 'operations' with current operation as first list element
2434 db_nslcmop_admin
.update({'operations': [new_op
]})
2435 op_list
= db_nslcmop_admin
.get('operations')
2437 # Existing operations, append operation to list
2438 op_list
.append(new_op
)
2440 db_nslcmop_update
= {'_admin.operations': op_list
}
2441 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2442 op_index
= len(op_list
) - 1
2445 # Helper methods for scale() sub-operations
2447 # pre-scale/post-scale:
2448 # Check for 3 different cases:
2449 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2450 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2451 # c. Reintent: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2452 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2453 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2454 # Find this sub-operation
2455 if (RO_nsr_id
and RO_scaling_info
):
2456 operationType
= 'SCALE-RO'
2458 'member_vnf_index': vnf_index
,
2459 'RO_nsr_id': RO_nsr_id
,
2460 'RO_scaling_info': RO_scaling_info
,
2464 'member_vnf_index': vnf_index
,
2465 'primitive': vnf_config_primitive
,
2466 'primitive_params': primitive_params
,
2467 'lcmOperationType': operationType
2469 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2470 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2471 # a. New sub-operation
2472 # The sub-operation does not exist, add it.
2473 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2474 # The following parameters are set to None for all kind of scaling:
2476 vdu_count_index
= None
2478 if RO_nsr_id
and RO_scaling_info
:
2479 vnf_config_primitive
= None
2480 primitive_params
= None
2483 RO_scaling_info
= None
2484 # Initial status for sub-operation
2485 operationState
= 'PROCESSING'
2486 detailed_status
= 'In progress'
2487 # Add sub-operation for pre/post-scaling (zero or more operations)
2488 self
._add
_suboperation
(db_nslcmop
,
2493 vnf_config_primitive
,
2500 return self
.SUBOPERATION_STATUS_NEW
2502 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2503 # or op_index (operationState != 'COMPLETED')
2504 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2506 # Function to return execution_environment id
2508 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2509 # TODO vdu_index_count
2510 for vca
in vca_deployed_list
:
2511 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2514 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
, vca_index
, destroy_ee
=True):
2516 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2517 :param logging_text:
2519 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2520 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2521 :param vca_index: index in the database _admin.deployed.VCA
2522 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2523 :return: None or exception
2525 # execute terminate_primitives
2526 terminate_primitives
= config_descriptor
.get("terminate-config-primitive")
2527 vdu_id
= vca_deployed
.get("vdu_id")
2528 vdu_count_index
= vca_deployed
.get("vdu_count_index")
2529 vdu_name
= vca_deployed
.get("vdu_name")
2530 vnf_index
= vca_deployed
.get("member-vnf-index")
2531 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
2532 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2533 terminate_primitives
= sorted(terminate_primitives
, key
=lambda x
: int(x
['seq']))
2534 for seq
in terminate_primitives
:
2535 # For each sequence in list, get primitive and call _ns_execute_primitive()
2536 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
2537 vnf_index
, seq
.get("name"))
2538 self
.logger
.debug(logging_text
+ step
)
2539 # Create the primitive for each sequence, i.e. "primitive": "touch"
2540 primitive
= seq
.get('name')
2541 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
2542 # The following 3 parameters are currently set to None for 'terminate':
2543 # vdu_id, vdu_count_index, vdu_name
2546 self
._add
_suboperation
(db_nslcmop
,
2552 mapped_primitive_params
)
2553 # Sub-operations: Call _ns_execute_primitive() instead of action()
2555 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
2556 mapped_primitive_params
)
2557 except LcmException
:
2558 # this happens when VCA is not deployed. In this case it is not needed to terminate
2560 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
2561 if result
not in result_ok
:
2562 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
2563 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
2564 # set that this VCA do not need terminated
2565 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
2566 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
2569 await self
.n2vc
.delete_execution_environment(vca_deployed
["ee_id"])
2571 async def _delete_all_N2VC(self
, db_nsr
: dict):
2572 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
2573 namespace
= "." + db_nsr
["_id"]
2575 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
2576 except N2VCNotFound
: # already deleted. Skip
2578 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
2580 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
2582 Terminates a deployment from RO
2583 :param logging_text:
2584 :param nsr_deployed: db_nsr._admin.deployed
2587 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
2588 this method will update only the index 2, but it will write on database the concatenated content of the list
2593 ro_nsr_id
= ro_delete_action
= None
2594 if nsr_deployed
and nsr_deployed
.get("RO"):
2595 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
2596 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
2599 stage
[2] = "Deleting ns from VIM."
2600 db_nsr_update
["detailed-status"] = " ".join(stage
)
2601 self
._write
_op
_status
(nslcmop_id
, stage
)
2602 self
.logger
.debug(logging_text
+ stage
[2])
2603 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2604 self
._write
_op
_status
(nslcmop_id
, stage
)
2605 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
2606 ro_delete_action
= desc
["action_id"]
2607 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
2608 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2609 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2610 if ro_delete_action
:
2611 # wait until NS is deleted from VIM
2612 stage
[2] = "Waiting ns deleted from VIM."
2613 detailed_status_old
= None
2614 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
2616 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2617 self
._write
_op
_status
(nslcmop_id
, stage
)
2619 delete_timeout
= 20 * 60 # 20 minutes
2620 while delete_timeout
> 0:
2621 desc
= await self
.RO
.show(
2623 item_id_name
=ro_nsr_id
,
2624 extra_item
="action",
2625 extra_item_id
=ro_delete_action
)
2628 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
2630 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
2631 if ns_status
== "ERROR":
2632 raise ROclient
.ROClientException(ns_status_info
)
2633 elif ns_status
== "BUILD":
2634 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
2635 elif ns_status
== "ACTIVE":
2636 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2637 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2640 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
2641 if stage
[2] != detailed_status_old
:
2642 detailed_status_old
= stage
[2]
2643 db_nsr_update
["detailed-status"] = " ".join(stage
)
2644 self
._write
_op
_status
(nslcmop_id
, stage
)
2645 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2646 await asyncio
.sleep(5, loop
=self
.loop
)
2648 else: # delete_timeout <= 0:
2649 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
2651 except Exception as e
:
2652 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2653 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2654 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2655 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2656 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2657 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
2658 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2659 failed_detail
.append("delete conflict: {}".format(e
))
2660 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
2662 failed_detail
.append("delete error: {}".format(e
))
2663 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
2666 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
2667 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
2669 stage
[2] = "Deleting nsd from RO."
2670 db_nsr_update
["detailed-status"] = " ".join(stage
)
2671 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2672 self
._write
_op
_status
(nslcmop_id
, stage
)
2673 await self
.RO
.delete("nsd", ro_nsd_id
)
2674 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
2675 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2676 except Exception as e
:
2677 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2678 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2679 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
2680 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2681 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
2682 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2684 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
2685 self
.logger
.error(logging_text
+ failed_detail
[-1])
2687 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
2688 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
2689 if not vnf_deployed
or not vnf_deployed
["id"]:
2692 ro_vnfd_id
= vnf_deployed
["id"]
2693 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
2694 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
2695 db_nsr_update
["detailed-status"] = " ".join(stage
)
2696 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2697 self
._write
_op
_status
(nslcmop_id
, stage
)
2698 await self
.RO
.delete("vnfd", ro_vnfd_id
)
2699 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
2700 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2701 except Exception as e
:
2702 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2703 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2704 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
2705 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2706 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
2707 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2709 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
2710 self
.logger
.error(logging_text
+ failed_detail
[-1])
2713 stage
[2] = "Error deleting from VIM"
2715 stage
[2] = "Deleted from VIM"
2716 db_nsr_update
["detailed-status"] = " ".join(stage
)
2717 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2718 self
._write
_op
_status
(nslcmop_id
, stage
)
2721 raise LcmException("; ".join(failed_detail
))
2723 async def terminate(self
, nsr_id
, nslcmop_id
):
2724 # Try to lock HA task here
2725 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
2726 if not task_is_locked_by_me
:
2729 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
2730 self
.logger
.debug(logging_text
+ "Enter")
2731 timeout_ns_terminate
= self
.timeout_ns_terminate
2735 error_list
= [] # annotates all failed error messages
2736 db_nslcmop_update
= {}
2737 autoremove
= False # autoremove after terminated
2738 tasks_dict_info
= {}
2740 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
2741 # ^ contains [stage, step, VIM-status]
2743 # wait for any previous tasks in process
2744 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
2746 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2747 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2748 operation_params
= db_nslcmop
.get("operationParams") or {}
2749 if operation_params
.get("timeout_ns_terminate"):
2750 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
2751 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2752 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2754 db_nsr_update
["operational-status"] = "terminating"
2755 db_nsr_update
["config-status"] = "terminating"
2756 self
._write
_ns
_status
(
2758 ns_state
="TERMINATING",
2759 current_operation
="TERMINATING",
2760 current_operation_id
=nslcmop_id
,
2761 other_update
=db_nsr_update
2763 self
._write
_op
_status
(
2768 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
2769 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
2772 stage
[1] = "Getting vnf descriptors from db."
2773 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2774 db_vnfds_from_id
= {}
2775 db_vnfds_from_member_index
= {}
2777 for vnfr
in db_vnfrs_list
:
2778 vnfd_id
= vnfr
["vnfd-id"]
2779 if vnfd_id
not in db_vnfds_from_id
:
2780 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2781 db_vnfds_from_id
[vnfd_id
] = vnfd
2782 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
2784 # Destroy individual execution environments when there are terminating primitives.
2785 # Rest of EE will be deleted at once
2786 if not operation_params
.get("skip_terminate_primitives"):
2787 stage
[0] = "Stage 2/3 execute terminating primitives."
2788 stage
[1] = "Looking execution environment that needs terminate."
2789 self
.logger
.debug(logging_text
+ stage
[1])
2790 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
2791 config_descriptor
= None
2792 if not vca
or not vca
.get("ee_id") or not vca
.get("needed_terminate"):
2794 if not vca
.get("member-vnf-index"):
2796 config_descriptor
= db_nsr
.get("ns-configuration")
2797 elif vca
.get("vdu_id"):
2798 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2799 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
2801 config_descriptor
= vdud
.get("vdu-configuration")
2802 elif vca
.get("kdu_name"):
2803 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2804 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
2806 config_descriptor
= kdud
.get("kdu-configuration")
2808 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
2809 task
= asyncio
.ensure_future(self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
,
2811 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
2813 # wait for pending tasks of terminate primitives
2815 self
.logger
.debug(logging_text
+ 'Waiting for terminate primitive pending tasks...')
2816 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
2817 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
2820 return # raise LcmException("; ".join(error_list))
2821 tasks_dict_info
.clear()
2823 # remove All execution environments at once
2824 stage
[0] = "Stage 3/3 delete all."
2826 if nsr_deployed
.get("VCA"):
2827 stage
[1] = "Deleting all execution environments."
2828 self
.logger
.debug(logging_text
+ stage
[1])
2829 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
2830 timeout
=self
.timeout_charm_delete
))
2831 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
2832 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
2834 # Delete from k8scluster
2835 stage
[1] = "Deleting KDUs."
2836 self
.logger
.debug(logging_text
+ stage
[1])
2837 # print(nsr_deployed)
2838 for kdu
in get_iterable(nsr_deployed
, "K8s"):
2839 if not kdu
or not kdu
.get("kdu-instance"):
2841 kdu_instance
= kdu
.get("kdu-instance")
2842 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
2843 task_delete_kdu_instance
= asyncio
.ensure_future(
2844 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
2845 cluster_uuid
=kdu
.get("k8scluster-uuid"),
2846 kdu_instance
=kdu_instance
))
2848 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
2849 format(kdu
.get("k8scluster-type")))
2851 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
2854 stage
[1] = "Deleting ns from VIM."
2855 task_delete_ro
= asyncio
.ensure_future(
2856 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
2857 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
2859 # rest of staff will be done at finally
2861 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2862 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
2864 except asyncio
.CancelledError
:
2865 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2866 exc
= "Operation was cancelled"
2867 except Exception as e
:
2868 exc
= traceback
.format_exc()
2869 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2872 error_list
.append(str(exc
))
2874 # wait for pending tasks
2876 stage
[1] = "Waiting for terminate pending tasks."
2877 self
.logger
.debug(logging_text
+ stage
[1])
2878 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
2880 stage
[1] = stage
[2] = ""
2881 except asyncio
.CancelledError
:
2882 error_list
.append("Cancelled")
2883 # TODO cancell all tasks
2884 except Exception as exc
:
2885 error_list
.append(str(exc
))
2886 # update status at database
2888 error_detail
= "; ".join(error_list
)
2889 # self.logger.error(logging_text + error_detail)
2890 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
2891 error_description_nsr
= 'Operation: TERMINATING.{}, Stage {}.'.format(nslcmop_id
, stage
[0])
2893 db_nsr_update
["operational-status"] = "failed"
2894 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2895 db_nslcmop_update
["detailed-status"] = error_detail
2896 nslcmop_operation_state
= "FAILED"
2900 error_description_nsr
= error_description_nslcmop
= None
2901 ns_state
= "NOT_INSTANTIATED"
2902 db_nsr_update
["operational-status"] = "terminated"
2903 db_nsr_update
["detailed-status"] = "Done"
2904 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
2905 db_nslcmop_update
["detailed-status"] = "Done"
2906 nslcmop_operation_state
= "COMPLETED"
2909 self
._write
_ns
_status
(
2912 current_operation
="IDLE",
2913 current_operation_id
=None,
2914 error_description
=error_description_nsr
,
2915 error_detail
=error_detail
,
2916 other_update
=db_nsr_update
2919 self
._write
_op
_status
(
2922 error_message
=error_description_nslcmop
,
2923 operation_state
=nslcmop_operation_state
,
2924 other_update
=db_nslcmop_update
,
2926 autoremove
= operation_params
.get("autoremove", False)
2927 if nslcmop_operation_state
:
2929 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2930 "operationState": nslcmop_operation_state
,
2931 "autoremove": autoremove
},
2933 except Exception as e
:
2934 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2936 self
.logger
.debug(logging_text
+ "Exit")
2937 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
2939 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
2941 error_detail_list
= []
2943 pending_tasks
= list(created_tasks_info
.keys())
2944 num_tasks
= len(pending_tasks
)
2946 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
2947 self
._write
_op
_status
(nslcmop_id
, stage
)
2948 while pending_tasks
:
2950 _timeout
= timeout
+ time_start
- time()
2951 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
2952 return_when
=asyncio
.FIRST_COMPLETED
)
2953 num_done
+= len(done
)
2954 if not done
: # Timeout
2955 for task
in pending_tasks
:
2956 new_error
= created_tasks_info
[task
] + ": Timeout"
2957 error_detail_list
.append(new_error
)
2958 error_list
.append(new_error
)
2961 if task
.cancelled():
2964 exc
= task
.exception()
2966 if isinstance(exc
, asyncio
.TimeoutError
):
2968 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
2969 error_list
.append(created_tasks_info
[task
])
2970 error_detail_list
.append(new_error
)
2971 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
2973 self
.logger
.error(logging_text
+ new_error
)
2975 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
2976 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
2978 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
2979 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
2981 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
2982 if nsr_id
: # update also nsr
2983 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
2984 "errorDetail": ". ".join(error_detail_list
)})
2985 self
._write
_op
_status
(nslcmop_id
, stage
)
2986 return error_detail_list
2989 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
2991 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
2992 The default-value is used. If it is between < > it look for a value at instantiation_params
2993 :param primitive_desc: portion of VNFD/NSD that describes primitive
2994 :param params: Params provided by user
2995 :param instantiation_params: Instantiation params provided by user
2996 :return: a dictionary with the calculated params
2998 calculated_params
= {}
2999 for parameter
in primitive_desc
.get("parameter", ()):
3000 param_name
= parameter
["name"]
3001 if param_name
in params
:
3002 calculated_params
[param_name
] = params
[param_name
]
3003 elif "default-value" in parameter
or "value" in parameter
:
3004 if "value" in parameter
:
3005 calculated_params
[param_name
] = parameter
["value"]
3007 calculated_params
[param_name
] = parameter
["default-value"]
3008 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3009 and calculated_params
[param_name
].endswith(">"):
3010 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3011 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3013 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3014 format(calculated_params
[param_name
], primitive_desc
["name"]))
3016 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3017 format(param_name
, primitive_desc
["name"]))
3019 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3020 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3022 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3023 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3025 # add always ns_config_info if primitive name is config
3026 if primitive_desc
["name"] == "config":
3027 if "ns_config_info" in instantiation_params
:
3028 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3029 return calculated_params
3031 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None):
3032 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3033 for vca
in deployed_vca
:
3036 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3038 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3040 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3044 # vca_deployed not found
3045 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} is not "
3046 "deployed".format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3049 ee_id
= vca
.get("ee_id")
3051 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3052 "execution environment"
3053 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3056 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3057 retries_interval
=30, timeout
=None) -> (str, str):
3059 if primitive
== "config":
3060 primitive_params
= {"params": primitive_params
}
3064 output
= await asyncio
.wait_for(
3065 self
.n2vc
.exec_primitive(
3067 primitive_name
=primitive
,
3068 params_dict
=primitive_params
,
3069 progress_timeout
=self
.timeout_progress_primitive
,
3070 total_timeout
=self
.timeout_primitive
),
3071 timeout
=timeout
or self
.timeout_primitive
)
3074 except asyncio
.CancelledError
:
3076 except Exception as e
: # asyncio.TimeoutError
3077 if isinstance(e
, asyncio
.TimeoutError
):
3081 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3083 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3085 return 'FAILED', str(e
)
3087 return 'COMPLETED', output
3089 except (LcmException
, asyncio
.CancelledError
):
3091 except Exception as e
:
3092 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3094 async def action(self
, nsr_id
, nslcmop_id
):
3096 # Try to lock HA task here
3097 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3098 if not task_is_locked_by_me
:
3101 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3102 self
.logger
.debug(logging_text
+ "Enter")
3103 # get all needed from database
3107 db_nslcmop_update
= {}
3108 nslcmop_operation_state
= None
3109 error_description_nslcmop
= None
3112 # wait for any previous tasks in process
3113 step
= "Waiting for previous operations to terminate"
3114 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3116 self
._write
_ns
_status
(
3119 current_operation
="RUNNING ACTION",
3120 current_operation_id
=nslcmop_id
3123 step
= "Getting information from database"
3124 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3125 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3127 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3128 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3129 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3130 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3131 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3132 primitive
= db_nslcmop
["operationParams"]["primitive"]
3133 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3134 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3137 step
= "Getting vnfr from database"
3138 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3139 step
= "Getting vnfd from database"
3140 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3142 step
= "Getting nsd from database"
3143 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3145 # for backward compatibility
3146 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3147 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3148 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3149 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3151 # look for primitive
3152 config_primitive_desc
= None
3154 for vdu
in get_iterable(db_vnfd
, "vdu"):
3155 if vdu_id
== vdu
["id"]:
3156 for config_primitive
in deep_get(vdu
, ("vdu-configuration", "config-primitive"), ()):
3157 if config_primitive
["name"] == primitive
:
3158 config_primitive_desc
= config_primitive
3162 for kdu
in get_iterable(db_vnfd
, "kdu"):
3163 if kdu_name
== kdu
["name"]:
3164 for config_primitive
in deep_get(kdu
, ("kdu-configuration", "config-primitive"), ()):
3165 if config_primitive
["name"] == primitive
:
3166 config_primitive_desc
= config_primitive
3170 for config_primitive
in deep_get(db_vnfd
, ("vnf-configuration", "config-primitive"), ()):
3171 if config_primitive
["name"] == primitive
:
3172 config_primitive_desc
= config_primitive
3175 for config_primitive
in deep_get(db_nsd
, ("ns-configuration", "config-primitive"), ()):
3176 if config_primitive
["name"] == primitive
:
3177 config_primitive_desc
= config_primitive
3180 if not config_primitive_desc
and not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3181 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3186 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3187 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3189 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3190 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3192 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3194 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3197 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3199 # TODO check if ns is in a proper status
3200 if kdu_name
and (primitive
in ("upgrade", "rollback", "status") or kdu_action
):
3201 # kdur and desc_params already set from before
3202 if primitive_params
:
3203 desc_params
.update(primitive_params
)
3204 # TODO Check if we will need something at vnf level
3205 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3206 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3209 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3211 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3212 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3213 raise LcmException(msg
)
3215 db_dict
= {"collection": "nsrs",
3216 "filter": {"_id": nsr_id
},
3217 "path": "_admin.deployed.K8s.{}".format(index
)}
3218 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive
, vnf_index
, kdu_name
))
3219 step
= "Executing kdu {}".format(primitive
)
3220 if primitive
== "upgrade":
3221 if desc_params
.get("kdu_model"):
3222 kdu_model
= desc_params
.get("kdu_model")
3223 del desc_params
["kdu_model"]
3225 kdu_model
= kdu
.get("kdu-model")
3226 parts
= kdu_model
.split(sep
=":")
3228 kdu_model
= parts
[0]
3230 detailed_status
= await asyncio
.wait_for(
3231 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3232 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3233 kdu_instance
=kdu
.get("kdu-instance"),
3234 atomic
=True, kdu_model
=kdu_model
,
3235 params
=desc_params
, db_dict
=db_dict
,
3236 timeout
=timeout_ns_action
),
3237 timeout
=timeout_ns_action
+ 10)
3238 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3239 elif primitive
== "rollback":
3240 detailed_status
= await asyncio
.wait_for(
3241 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3242 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3243 kdu_instance
=kdu
.get("kdu-instance"),
3245 timeout
=timeout_ns_action
)
3246 elif primitive
== "status":
3247 detailed_status
= await asyncio
.wait_for(
3248 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3249 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3250 kdu_instance
=kdu
.get("kdu-instance")),
3251 timeout
=timeout_ns_action
)
3253 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3254 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3256 detailed_status
= await asyncio
.wait_for(
3257 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3258 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3259 kdu_instance
=kdu_instance
,
3260 primitive_name
=primitive
,
3261 params
=params
, db_dict
=db_dict
,
3262 timeout
=timeout_ns_action
),
3263 timeout
=timeout_ns_action
)
3266 nslcmop_operation_state
= 'COMPLETED'
3268 detailed_status
= ''
3269 nslcmop_operation_state
= 'FAILED'
3271 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3272 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3273 member_vnf_index
=vnf_index
,
3275 vdu_count_index
=vdu_count_index
),
3276 primitive
=primitive
,
3277 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3278 timeout
=timeout_ns_action
)
3280 db_nslcmop_update
["detailed-status"] = detailed_status
3281 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3282 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3284 return # database update is called inside finally
3286 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3287 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3289 except asyncio
.CancelledError
:
3290 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3291 exc
= "Operation was cancelled"
3292 except asyncio
.TimeoutError
:
3293 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3295 except Exception as e
:
3296 exc
= traceback
.format_exc()
3297 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3300 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3301 "FAILED {}: {}".format(step
, exc
)
3302 nslcmop_operation_state
= "FAILED"
3304 self
._write
_ns
_status
(
3306 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3307 current_operation
="IDLE",
3308 current_operation_id
=None,
3309 # error_description=error_description_nsr,
3310 # error_detail=error_detail,
3311 other_update
=db_nsr_update
3315 self
._write
_op
_status
(
3318 error_message
=error_description_nslcmop
,
3319 operation_state
=nslcmop_operation_state
,
3320 other_update
=db_nslcmop_update
,
3323 if nslcmop_operation_state
:
3325 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3326 "operationState": nslcmop_operation_state
},
3328 except Exception as e
:
3329 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3330 self
.logger
.debug(logging_text
+ "Exit")
3331 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3332 return nslcmop_operation_state
, detailed_status
3334 async def scale(self
, nsr_id
, nslcmop_id
):
3336 # Try to lock HA task here
3337 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3338 if not task_is_locked_by_me
:
3341 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3342 self
.logger
.debug(logging_text
+ "Enter")
3343 # get all needed from database
3346 db_nslcmop_update
= {}
3347 nslcmop_operation_state
= None
3350 # in case of error, indicates what part of scale was failed to put nsr at error status
3351 scale_process
= None
3352 old_operational_status
= ""
3353 old_config_status
= ""
3356 # wait for any previous tasks in process
3357 step
= "Waiting for previous operations to terminate"
3358 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3360 self
._write
_ns
_status
(
3363 current_operation
="SCALING",
3364 current_operation_id
=nslcmop_id
3367 step
= "Getting nslcmop from database"
3368 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3369 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3370 step
= "Getting nsr from database"
3371 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3373 old_operational_status
= db_nsr
["operational-status"]
3374 old_config_status
= db_nsr
["config-status"]
3375 step
= "Parsing scaling parameters"
3376 # self.logger.debug(step)
3377 db_nsr_update
["operational-status"] = "scaling"
3378 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3379 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3382 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3383 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3384 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3385 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3386 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3389 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3390 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3391 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3392 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3393 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3395 # for backward compatibility
3396 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3397 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3398 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3399 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3401 step
= "Getting vnfr from database"
3402 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3403 step
= "Getting vnfd from database"
3404 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3406 step
= "Getting scaling-group-descriptor"
3407 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3408 if scaling_descriptor
["name"] == scaling_group
:
3411 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3412 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3415 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3416 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3417 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3420 # TODO check if ns is in a proper status
3421 step
= "Sending scale order to VIM"
3423 if not db_nsr
["_admin"].get("scaling-group"):
3424 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3425 admin_scale_index
= 0
3427 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3428 if admin_scale_info
["name"] == scaling_group
:
3429 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3431 else: # not found, set index one plus last element and add new entry with the name
3432 admin_scale_index
+= 1
3433 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3434 RO_scaling_info
= []
3435 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3436 if scaling_type
== "SCALE_OUT":
3437 # count if max-instance-count is reached
3438 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3439 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3440 if nb_scale_op
>= max_instance_count
:
3441 raise LcmException("reached the limit of {} (max-instance-count) "
3442 "scaling-out operations for the "
3443 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3446 vdu_scaling_info
["scaling_direction"] = "OUT"
3447 vdu_scaling_info
["vdu-create"] = {}
3448 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3449 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3450 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3451 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3453 elif scaling_type
== "SCALE_IN":
3454 # count if min-instance-count is reached
3455 min_instance_count
= 0
3456 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3457 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3458 if nb_scale_op
<= min_instance_count
:
3459 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3460 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3462 vdu_scaling_info
["scaling_direction"] = "IN"
3463 vdu_scaling_info
["vdu-delete"] = {}
3464 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3465 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3466 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3467 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3469 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3470 vdu_create
= vdu_scaling_info
.get("vdu-create")
3471 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3472 if vdu_scaling_info
["scaling_direction"] == "IN":
3473 for vdur
in reversed(db_vnfr
["vdur"]):
3474 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3475 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3476 vdu_scaling_info
["vdu"].append({
3477 "name": vdur
["name"],
3478 "vdu_id": vdur
["vdu-id-ref"],
3481 for interface
in vdur
["interfaces"]:
3482 vdu_scaling_info
["vdu"][-1]["interface"].append({
3483 "name": interface
["name"],
3484 "ip_address": interface
["ip-address"],
3485 "mac_address": interface
.get("mac-address"),
3487 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
3490 step
= "Executing pre-scale vnf-config-primitive"
3491 if scaling_descriptor
.get("scaling-config-action"):
3492 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3493 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
3494 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
3495 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3496 step
= db_nslcmop_update
["detailed-status"] = \
3497 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3499 # look for primitive
3500 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3501 if config_primitive
["name"] == vnf_config_primitive
:
3505 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
3506 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
3507 "primitive".format(scaling_group
, config_primitive
))
3509 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3510 if db_vnfr
.get("additionalParamsForVnf"):
3511 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3513 scale_process
= "VCA"
3514 db_nsr_update
["config-status"] = "configuring pre-scaling"
3515 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3517 # Pre-scale reintent check: Check if this sub-operation has been executed before
3518 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3519 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
3520 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3521 # Skip sub-operation
3522 result
= 'COMPLETED'
3523 result_detail
= 'Done'
3524 self
.logger
.debug(logging_text
+
3525 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
3526 vnf_config_primitive
, result
, result_detail
))
3528 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3529 # New sub-operation: Get index of this sub-operation
3530 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3531 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3532 format(vnf_config_primitive
))
3534 # Reintent: Get registered params for this existing sub-operation
3535 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3536 vnf_index
= op
.get('member_vnf_index')
3537 vnf_config_primitive
= op
.get('primitive')
3538 primitive_params
= op
.get('primitive_params')
3539 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3540 format(vnf_config_primitive
))
3541 # Execute the primitive, either with new (first-time) or registered (reintent) args
3542 result
, result_detail
= await self
._ns
_execute
_primitive
(
3543 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3544 member_vnf_index
=vnf_index
,
3546 vdu_count_index
=None),
3547 vnf_config_primitive
, primitive_params
)
3548 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3549 vnf_config_primitive
, result
, result_detail
))
3550 # Update operationState = COMPLETED | FAILED
3551 self
._update
_suboperation
_status
(
3552 db_nslcmop
, op_index
, result
, result_detail
)
3554 if result
== "FAILED":
3555 raise LcmException(result_detail
)
3556 db_nsr_update
["config-status"] = old_config_status
3557 scale_process
= None
3561 # Should this block be skipped if 'RO_nsr_id' == None ?
3562 # if (RO_nsr_id and RO_scaling_info):
3564 scale_process
= "RO"
3565 # Scale RO reintent check: Check if this sub-operation has been executed before
3566 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3567 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
3568 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3569 # Skip sub-operation
3570 result
= 'COMPLETED'
3571 result_detail
= 'Done'
3572 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
3573 result
, result_detail
))
3575 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3576 # New sub-operation: Get index of this sub-operation
3577 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3578 self
.logger
.debug(logging_text
+ "New sub-operation RO")
3580 # Reintent: Get registered params for this existing sub-operation
3581 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3582 RO_nsr_id
= op
.get('RO_nsr_id')
3583 RO_scaling_info
= op
.get('RO_scaling_info')
3584 self
.logger
.debug(logging_text
+ "Sub-operation RO reintent".format(
3585 vnf_config_primitive
))
3587 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
3588 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
3589 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
3591 RO_nslcmop_id
= RO_desc
["instance_action_id"]
3592 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
3594 RO_task_done
= False
3595 step
= detailed_status
= "Waiting RO_task_id={} to complete the scale action.".format(RO_nslcmop_id
)
3596 detailed_status_old
= None
3597 self
.logger
.debug(logging_text
+ step
)
3599 deployment_timeout
= 1 * 3600 # One hour
3600 while deployment_timeout
> 0:
3601 if not RO_task_done
:
3602 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
3603 extra_item_id
=RO_nslcmop_id
)
3606 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3608 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3609 if ns_status
== "ERROR":
3610 raise ROclient
.ROClientException(ns_status_info
)
3611 elif ns_status
== "BUILD":
3612 detailed_status
= step
+ "; {}".format(ns_status_info
)
3613 elif ns_status
== "ACTIVE":
3615 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
3616 self
.logger
.debug(logging_text
+ step
)
3618 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3621 if ns_status
== "ERROR":
3622 raise ROclient
.ROClientException(ns_status_info
)
3623 elif ns_status
== "BUILD":
3624 detailed_status
= step
+ "; {}".format(ns_status_info
)
3625 elif ns_status
== "ACTIVE":
3626 step
= detailed_status
= \
3627 "Waiting for management IP address reported by the VIM. Updating VNFRs"
3629 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
3632 desc
= await self
.RO
.show("ns", RO_nsr_id
)
3635 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3637 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
3638 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
3640 except LcmExceptionNoMgmtIP
:
3643 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
3644 if detailed_status
!= detailed_status_old
:
3645 self
._update
_suboperation
_status
(
3646 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
3647 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
3648 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3650 await asyncio
.sleep(5, loop
=self
.loop
)
3651 deployment_timeout
-= 5
3652 if deployment_timeout
<= 0:
3653 self
._update
_suboperation
_status
(
3654 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
3655 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
3657 # update VDU_SCALING_INFO with the obtained ip_addresses
3658 if vdu_scaling_info
["scaling_direction"] == "OUT":
3659 for vdur
in reversed(db_vnfr
["vdur"]):
3660 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
3661 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
3662 vdu_scaling_info
["vdu"].append({
3663 "name": vdur
["name"],
3664 "vdu_id": vdur
["vdu-id-ref"],
3667 for interface
in vdur
["interfaces"]:
3668 vdu_scaling_info
["vdu"][-1]["interface"].append({
3669 "name": interface
["name"],
3670 "ip_address": interface
["ip-address"],
3671 "mac_address": interface
.get("mac-address"),
3673 del vdu_scaling_info
["vdu-create"]
3675 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
3678 scale_process
= None
3680 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3683 # execute primitive service POST-SCALING
3684 step
= "Executing post-scale vnf-config-primitive"
3685 if scaling_descriptor
.get("scaling-config-action"):
3686 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3687 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
3688 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
3689 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3690 step
= db_nslcmop_update
["detailed-status"] = \
3691 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3693 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3694 if db_vnfr
.get("additionalParamsForVnf"):
3695 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3697 # look for primitive
3698 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3699 if config_primitive
["name"] == vnf_config_primitive
:
3702 raise LcmException("Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:"
3703 "scaling-config-action[vnf-config-primitive-name-ref='{}'] does not "
3704 "match any vnf-configuration:config-primitive".format(scaling_group
,
3706 scale_process
= "VCA"
3707 db_nsr_update
["config-status"] = "configuring post-scaling"
3708 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3710 # Post-scale reintent check: Check if this sub-operation has been executed before
3711 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3712 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
3713 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3714 # Skip sub-operation
3715 result
= 'COMPLETED'
3716 result_detail
= 'Done'
3717 self
.logger
.debug(logging_text
+
3718 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
3719 format(vnf_config_primitive
, result
, result_detail
))
3721 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3722 # New sub-operation: Get index of this sub-operation
3723 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3724 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3725 format(vnf_config_primitive
))
3727 # Reintent: Get registered params for this existing sub-operation
3728 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3729 vnf_index
= op
.get('member_vnf_index')
3730 vnf_config_primitive
= op
.get('primitive')
3731 primitive_params
= op
.get('primitive_params')
3732 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3733 format(vnf_config_primitive
))
3734 # Execute the primitive, either with new (first-time) or registered (reintent) args
3735 result
, result_detail
= await self
._ns
_execute
_primitive
(
3736 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3737 member_vnf_index
=vnf_index
,
3739 vdu_count_index
=None),
3740 vnf_config_primitive
, primitive_params
)
3741 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3742 vnf_config_primitive
, result
, result_detail
))
3743 # Update operationState = COMPLETED | FAILED
3744 self
._update
_suboperation
_status
(
3745 db_nslcmop
, op_index
, result
, result_detail
)
3747 if result
== "FAILED":
3748 raise LcmException(result_detail
)
3749 db_nsr_update
["config-status"] = old_config_status
3750 scale_process
= None
3753 db_nslcmop_update
["operationState"] = nslcmop_operation_state
= "COMPLETED"
3754 db_nslcmop_update
["statusEnteredTime"] = time()
3755 db_nslcmop_update
["detailed-status"] = "done"
3756 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
3757 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
3758 else old_operational_status
3759 db_nsr_update
["config-status"] = old_config_status
3761 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
3762 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3764 except asyncio
.CancelledError
:
3765 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3766 exc
= "Operation was cancelled"
3767 except Exception as e
:
3768 exc
= traceback
.format_exc()
3769 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3771 self
._write
_ns
_status
(
3774 current_operation
="IDLE",
3775 current_operation_id
=None
3779 db_nslcmop_update
["detailed-status"] = "FAILED {}: {}".format(step
, exc
)
3780 db_nslcmop_update
["operationState"] = nslcmop_operation_state
= "FAILED"
3781 db_nslcmop_update
["statusEnteredTime"] = time()
3783 db_nsr_update
["operational-status"] = old_operational_status
3784 db_nsr_update
["config-status"] = old_config_status
3785 db_nsr_update
["detailed-status"] = ""
3787 if "VCA" in scale_process
:
3788 db_nsr_update
["config-status"] = "failed"
3789 if "RO" in scale_process
:
3790 db_nsr_update
["operational-status"] = "failed"
3791 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
3794 if db_nslcmop
and db_nslcmop_update
:
3795 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3797 self
._write
_ns
_status
(
3800 current_operation
="IDLE",
3801 current_operation_id
=None,
3802 other_update
=db_nsr_update
3805 except DbException
as e
:
3806 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
3807 if nslcmop_operation_state
:
3809 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3810 "operationState": nslcmop_operation_state
},
3813 # await asyncio.sleep(cooldown_time, loop=self.loop)
3814 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
3815 except Exception as e
:
3816 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3817 self
.logger
.debug(logging_text
+ "Exit")
3818 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")