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
43 __author__
= "Alfonso Tierno"
47 timeout_vca_on_error
= 5 * 60 # Time for charm from first time at blocked,error status to mark as failed
48 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
49 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
50 timeout_charm_delete
= 10 * 60
51 timeout_primitive
= 10 * 60 # timeout for primitive execution
52 timeout_progress_primitive
= 2 * 60 # timeout for some progress in a primitive execution
54 SUBOPERATION_STATUS_NOT_FOUND
= -1
55 SUBOPERATION_STATUS_NEW
= -2
56 SUBOPERATION_STATUS_SKIP
= -3
57 task_name_deploy_vca
= "Deploying VCA"
59 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
61 Init, Connect to database, filesystem storage, and messaging
62 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
69 logger
=logging
.getLogger('lcm.ns')
73 self
.lcm_tasks
= lcm_tasks
74 self
.timeout
= config
["timeout"]
75 self
.ro_config
= config
["ro_config"]
76 self
.vca_config
= config
["VCA"].copy()
78 # create N2VC connector
79 self
.n2vc
= N2VCJujuConnector(
84 url
='{}:{}'.format(self
.vca_config
['host'], self
.vca_config
['port']),
85 username
=self
.vca_config
.get('user', None),
86 vca_config
=self
.vca_config
,
87 on_update_db
=self
._on
_update
_n
2vc
_db
90 self
.k8sclusterhelm
= K8sHelmConnector(
91 kubectl_command
=self
.vca_config
.get("kubectlpath"),
92 helm_command
=self
.vca_config
.get("helmpath"),
99 self
.k8sclusterjuju
= K8sJujuConnector(
100 kubectl_command
=self
.vca_config
.get("kubectlpath"),
101 juju_command
=self
.vca_config
.get("jujupath"),
108 self
.k8scluster_map
= {
109 "helm-chart": self
.k8sclusterhelm
,
110 "chart": self
.k8sclusterhelm
,
111 "juju-bundle": self
.k8sclusterjuju
,
112 "juju": self
.k8sclusterjuju
,
115 self
.RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
117 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
119 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
122 # TODO filter RO descriptor fields...
126 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
127 db_dict
['deploymentStatus'] = ro_descriptor
128 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
130 except Exception as e
:
131 self
.logger
.warn('Cannot write database RO deployment for ns={} -> {}'.format(nsrs_id
, e
))
133 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
):
135 # remove last dot from path (if exists)
136 if path
.endswith('.'):
139 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
140 # .format(table, filter, path, updated_data))
144 nsr_id
= filter.get('_id')
146 # read ns record from database
147 nsr
= self
.db
.get_one(table
='nsrs', q_filter
=filter)
148 current_ns_status
= nsr
.get('nsState')
150 # get vca status for NS
151 status_dict
= await self
.n2vc
.get_status(namespace
='.' + nsr_id
, yaml_format
=False)
155 db_dict
['vcaStatus'] = status_dict
157 # update configurationStatus for this VCA
159 vca_index
= int(path
[path
.rfind(".")+1:])
161 vca_list
= deep_get(target_dict
=nsr
, key_list
=('_admin', 'deployed', 'VCA'))
162 vca_status
= vca_list
[vca_index
].get('status')
164 configuration_status_list
= nsr
.get('configurationStatus')
165 config_status
= configuration_status_list
[vca_index
].get('status')
167 if config_status
== 'BROKEN' and vca_status
!= 'failed':
168 db_dict
['configurationStatus'][vca_index
] = 'READY'
169 elif config_status
!= 'BROKEN' and vca_status
== 'failed':
170 db_dict
['configurationStatus'][vca_index
] = 'BROKEN'
171 except Exception as e
:
172 # not update configurationStatus
173 self
.logger
.debug('Error updating vca_index (ignore): {}'.format(e
))
175 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
176 # if nsState = 'DEGRADED' check if all is OK
178 if current_ns_status
in ('READY', 'DEGRADED'):
179 error_description
= ''
181 if status_dict
.get('machines'):
182 for machine_id
in status_dict
.get('machines'):
183 machine
= status_dict
.get('machines').get(machine_id
)
184 # check machine agent-status
185 if machine
.get('agent-status'):
186 s
= machine
.get('agent-status').get('status')
189 error_description
+= 'machine {} agent-status={} ; '.format(machine_id
, s
)
190 # check machine instance status
191 if machine
.get('instance-status'):
192 s
= machine
.get('instance-status').get('status')
195 error_description
+= 'machine {} instance-status={} ; '.format(machine_id
, s
)
197 if status_dict
.get('applications'):
198 for app_id
in status_dict
.get('applications'):
199 app
= status_dict
.get('applications').get(app_id
)
200 # check application status
201 if app
.get('status'):
202 s
= app
.get('status').get('status')
205 error_description
+= 'application {} status={} ; '.format(app_id
, s
)
207 if error_description
:
208 db_dict
['errorDescription'] = error_description
209 if current_ns_status
== 'READY' and is_degraded
:
210 db_dict
['nsState'] = 'DEGRADED'
211 if current_ns_status
== 'DEGRADED' and not is_degraded
:
212 db_dict
['nsState'] = 'READY'
215 self
.update_db_2("nsrs", nsr_id
, db_dict
)
217 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
219 except Exception as e
:
220 self
.logger
.warn('Error updating NS state for ns={}: {}'.format(nsr_id
, e
))
222 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
224 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
225 :param vnfd: input vnfd
226 :param new_id: overrides vnf id if provided
227 :param additionalParams: Instantiation params for VNFs provided
228 :param nsrId: Id of the NSR
229 :return: copy of vnfd
232 vnfd_RO
= deepcopy(vnfd
)
233 # remove unused by RO configuration, monitoring, scaling and internal keys
234 vnfd_RO
.pop("_id", None)
235 vnfd_RO
.pop("_admin", None)
236 vnfd_RO
.pop("vnf-configuration", None)
237 vnfd_RO
.pop("monitoring-param", None)
238 vnfd_RO
.pop("scaling-group-descriptor", None)
239 vnfd_RO
.pop("kdu", None)
240 vnfd_RO
.pop("k8s-cluster", None)
242 vnfd_RO
["id"] = new_id
244 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
245 for vdu
in get_iterable(vnfd_RO
, "vdu"):
246 cloud_init_file
= None
247 if vdu
.get("cloud-init-file"):
248 base_folder
= vnfd
["_admin"]["storage"]
249 cloud_init_file
= "{}/{}/cloud_init/{}".format(base_folder
["folder"], base_folder
["pkg-dir"],
250 vdu
["cloud-init-file"])
251 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
252 cloud_init_content
= ci_file
.read()
253 vdu
.pop("cloud-init-file", None)
254 elif vdu
.get("cloud-init"):
255 cloud_init_content
= vdu
["cloud-init"]
260 ast
= env
.parse(cloud_init_content
)
261 mandatory_vars
= meta
.find_undeclared_variables(ast
)
263 for var
in mandatory_vars
:
264 if not additionalParams
or var
not in additionalParams
.keys():
265 raise LcmException("Variable '{}' defined at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
266 "file, must be provided in the instantiation parameters inside the "
267 "'additionalParamsForVnf' block".format(var
, vnfd
["id"], vdu
["id"]))
268 template
= Template(cloud_init_content
)
269 cloud_init_content
= template
.render(additionalParams
or {})
270 vdu
["cloud-init"] = cloud_init_content
273 except FsException
as e
:
274 raise LcmException("Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".
275 format(vnfd
["id"], vdu
["id"], cloud_init_file
, e
))
276 except (TemplateError
, TemplateNotFound
, TemplateSyntaxError
) as e
:
277 raise LcmException("Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".
278 format(vnfd
["id"], vdu
["id"], e
))
280 def ns_params_2_RO(self
, ns_params
, nsd
, vnfd_dict
, n2vc_key_list
):
282 Creates a RO ns descriptor from OSM ns_instantiate params
283 :param ns_params: OSM instantiate params
284 :return: The RO ns descriptor
288 # TODO feature 1417: Check that no instantiation is set over PDU
289 # check if PDU forces a concrete vim-network-id and add it
290 # check if PDU contains a SDN-assist info (dpid, switch, port) and pass it to RO
292 def vim_account_2_RO(vim_account
):
293 if vim_account
in vim_2_RO
:
294 return vim_2_RO
[vim_account
]
296 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
297 if db_vim
["_admin"]["operationalState"] != "ENABLED":
298 raise LcmException("VIM={} is not available. operationalState={}".format(
299 vim_account
, db_vim
["_admin"]["operationalState"]))
300 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
301 vim_2_RO
[vim_account
] = RO_vim_id
304 def wim_account_2_RO(wim_account
):
305 if isinstance(wim_account
, str):
306 if wim_account
in wim_2_RO
:
307 return wim_2_RO
[wim_account
]
309 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
310 if db_wim
["_admin"]["operationalState"] != "ENABLED":
311 raise LcmException("WIM={} is not available. operationalState={}".format(
312 wim_account
, db_wim
["_admin"]["operationalState"]))
313 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
314 wim_2_RO
[wim_account
] = RO_wim_id
319 def ip_profile_2_RO(ip_profile
):
320 RO_ip_profile
= deepcopy((ip_profile
))
321 if "dns-server" in RO_ip_profile
:
322 if isinstance(RO_ip_profile
["dns-server"], list):
323 RO_ip_profile
["dns-address"] = []
324 for ds
in RO_ip_profile
.pop("dns-server"):
325 RO_ip_profile
["dns-address"].append(ds
['address'])
327 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
328 if RO_ip_profile
.get("ip-version") == "ipv4":
329 RO_ip_profile
["ip-version"] = "IPv4"
330 if RO_ip_profile
.get("ip-version") == "ipv6":
331 RO_ip_profile
["ip-version"] = "IPv6"
332 if "dhcp-params" in RO_ip_profile
:
333 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
339 # "name": ns_params["nsName"],
340 # "description": ns_params.get("nsDescription"),
341 "datacenter": vim_account_2_RO(ns_params
["vimAccountId"]),
342 "wim_account": wim_account_2_RO(ns_params
.get("wimAccountId")),
343 # "scenario": ns_params["nsdId"],
346 n2vc_key_list
= n2vc_key_list
or []
347 for vnfd_ref
, vnfd
in vnfd_dict
.items():
348 vdu_needed_access
= []
350 if vnfd
.get("vnf-configuration"):
351 ssh_required
= deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required"))
352 if ssh_required
and vnfd
.get("mgmt-interface"):
353 if vnfd
["mgmt-interface"].get("vdu-id"):
354 vdu_needed_access
.append(vnfd
["mgmt-interface"]["vdu-id"])
355 elif vnfd
["mgmt-interface"].get("cp"):
356 mgmt_cp
= vnfd
["mgmt-interface"]["cp"]
358 for vdu
in vnfd
.get("vdu", ()):
359 if vdu
.get("vdu-configuration"):
360 ssh_required
= deep_get(vdu
, ("vdu-configuration", "config-access", "ssh-access", "required"))
362 vdu_needed_access
.append(vdu
["id"])
364 for vdu_interface
in vdu
.get("interface"):
365 if vdu_interface
.get("external-connection-point-ref") and \
366 vdu_interface
["external-connection-point-ref"] == mgmt_cp
:
367 vdu_needed_access
.append(vdu
["id"])
371 if vdu_needed_access
:
372 for vnf_member
in nsd
.get("constituent-vnfd"):
373 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
375 for vdu
in vdu_needed_access
:
376 populate_dict(RO_ns_params
,
377 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
, "mgmt_keys"),
380 if ns_params
.get("vduImage"):
381 RO_ns_params
["vduImage"] = ns_params
["vduImage"]
383 if ns_params
.get("ssh_keys"):
384 RO_ns_params
["cloud-config"] = {"key-pairs": ns_params
["ssh_keys"]}
385 for vnf_params
in get_iterable(ns_params
, "vnf"):
386 for constituent_vnfd
in nsd
["constituent-vnfd"]:
387 if constituent_vnfd
["member-vnf-index"] == vnf_params
["member-vnf-index"]:
388 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
391 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index={} is not present at nsd:"
392 "constituent-vnfd".format(vnf_params
["member-vnf-index"]))
393 if vnf_params
.get("vimAccountId"):
394 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "datacenter"),
395 vim_account_2_RO(vnf_params
["vimAccountId"]))
397 for vdu_params
in get_iterable(vnf_params
, "vdu"):
398 # TODO feature 1417: check that this VDU exist and it is not a PDU
399 if vdu_params
.get("volume"):
400 for volume_params
in vdu_params
["volume"]:
401 if volume_params
.get("vim-volume-id"):
402 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
403 vdu_params
["id"], "devices", volume_params
["name"], "vim_id"),
404 volume_params
["vim-volume-id"])
405 if vdu_params
.get("interface"):
406 for interface_params
in vdu_params
["interface"]:
407 if interface_params
.get("ip-address"):
408 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
409 vdu_params
["id"], "interfaces", interface_params
["name"],
411 interface_params
["ip-address"])
412 if interface_params
.get("mac-address"):
413 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
414 vdu_params
["id"], "interfaces", interface_params
["name"],
416 interface_params
["mac-address"])
417 if interface_params
.get("floating-ip-required"):
418 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
419 vdu_params
["id"], "interfaces", interface_params
["name"],
421 interface_params
["floating-ip-required"])
423 for internal_vld_params
in get_iterable(vnf_params
, "internal-vld"):
424 if internal_vld_params
.get("vim-network-name"):
425 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
426 internal_vld_params
["name"], "vim-network-name"),
427 internal_vld_params
["vim-network-name"])
428 if internal_vld_params
.get("vim-network-id"):
429 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
430 internal_vld_params
["name"], "vim-network-id"),
431 internal_vld_params
["vim-network-id"])
432 if internal_vld_params
.get("ip-profile"):
433 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
434 internal_vld_params
["name"], "ip-profile"),
435 ip_profile_2_RO(internal_vld_params
["ip-profile"]))
436 if internal_vld_params
.get("provider-network"):
438 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
439 internal_vld_params
["name"], "provider-network"),
440 internal_vld_params
["provider-network"].copy())
442 for icp_params
in get_iterable(internal_vld_params
, "internal-connection-point"):
445 for vdu_descriptor
in vnf_descriptor
["vdu"]:
446 for vdu_interface
in vdu_descriptor
["interface"]:
447 if vdu_interface
.get("internal-connection-point-ref") == icp_params
["id-ref"]:
448 if icp_params
.get("ip-address"):
449 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
450 vdu_descriptor
["id"], "interfaces",
451 vdu_interface
["name"], "ip_address"),
452 icp_params
["ip-address"])
454 if icp_params
.get("mac-address"):
455 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
456 vdu_descriptor
["id"], "interfaces",
457 vdu_interface
["name"], "mac_address"),
458 icp_params
["mac-address"])
464 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index[{}]:"
465 "internal-vld:id-ref={} is not present at vnfd:internal-"
466 "connection-point".format(vnf_params
["member-vnf-index"],
467 icp_params
["id-ref"]))
469 for vld_params
in get_iterable(ns_params
, "vld"):
470 if "ip-profile" in vld_params
:
471 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "ip-profile"),
472 ip_profile_2_RO(vld_params
["ip-profile"]))
474 if vld_params
.get("provider-network"):
476 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "provider-network"),
477 vld_params
["provider-network"].copy())
479 if "wimAccountId" in vld_params
and vld_params
["wimAccountId"] is not None:
480 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "wim_account"),
481 wim_account_2_RO(vld_params
["wimAccountId"])),
482 if vld_params
.get("vim-network-name"):
484 if isinstance(vld_params
["vim-network-name"], dict):
485 for vim_account
, vim_net
in vld_params
["vim-network-name"].items():
486 RO_vld_sites
.append({
487 "netmap-use": vim_net
,
488 "datacenter": vim_account_2_RO(vim_account
)
490 else: # isinstance str
491 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-name"]})
493 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
495 if vld_params
.get("vim-network-id"):
497 if isinstance(vld_params
["vim-network-id"], dict):
498 for vim_account
, vim_net
in vld_params
["vim-network-id"].items():
499 RO_vld_sites
.append({
500 "netmap-use": vim_net
,
501 "datacenter": vim_account_2_RO(vim_account
)
503 else: # isinstance str
504 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-id"]})
506 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
507 if vld_params
.get("ns-net"):
508 if isinstance(vld_params
["ns-net"], dict):
509 for vld_id
, instance_scenario_id
in vld_params
["ns-net"].items():
510 RO_vld_ns_net
= {"instance_scenario_id": instance_scenario_id
, "osm_id": vld_id
}
511 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "use-network"), RO_vld_ns_net
)
512 if "vnfd-connection-point-ref" in vld_params
:
513 for cp_params
in vld_params
["vnfd-connection-point-ref"]:
515 for constituent_vnfd
in nsd
["constituent-vnfd"]:
516 if constituent_vnfd
["member-vnf-index"] == cp_params
["member-vnf-index-ref"]:
517 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
521 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={} "
522 "is not present at nsd:constituent-vnfd".format(cp_params
["member-vnf-index-ref"]))
524 for vdu_descriptor
in vnf_descriptor
["vdu"]:
525 for interface_descriptor
in vdu_descriptor
["interface"]:
526 if interface_descriptor
.get("external-connection-point-ref") == \
527 cp_params
["vnfd-connection-point-ref"]:
534 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={}:"
535 "vnfd-connection-point-ref={} is not present at vnfd={}".format(
536 cp_params
["member-vnf-index-ref"],
537 cp_params
["vnfd-connection-point-ref"],
538 vnf_descriptor
["id"]))
539 if cp_params
.get("ip-address"):
540 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
541 vdu_descriptor
["id"], "interfaces",
542 interface_descriptor
["name"], "ip_address"),
543 cp_params
["ip-address"])
544 if cp_params
.get("mac-address"):
545 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
546 vdu_descriptor
["id"], "interfaces",
547 interface_descriptor
["name"], "mac_address"),
548 cp_params
["mac-address"])
551 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None):
552 # make a copy to do not change
553 vdu_create
= copy(vdu_create
)
554 vdu_delete
= copy(vdu_delete
)
556 vdurs
= db_vnfr
.get("vdur")
559 vdu_index
= len(vdurs
)
562 vdur
= vdurs
[vdu_index
]
563 if vdur
.get("pdu-type"):
565 vdu_id_ref
= vdur
["vdu-id-ref"]
566 if vdu_create
and vdu_create
.get(vdu_id_ref
):
567 for index
in range(0, vdu_create
[vdu_id_ref
]):
568 vdur
= deepcopy(vdur
)
569 vdur
["_id"] = str(uuid4())
570 vdur
["count-index"] += 1
571 vdurs
.insert(vdu_index
+1+index
, vdur
)
572 del vdu_create
[vdu_id_ref
]
573 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
575 vdu_delete
[vdu_id_ref
] -= 1
576 if not vdu_delete
[vdu_id_ref
]:
577 del vdu_delete
[vdu_id_ref
]
578 # check all operations are done
579 if vdu_create
or vdu_delete
:
580 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
583 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
586 vnfr_update
= {"vdur": vdurs
}
587 db_vnfr
["vdur"] = vdurs
588 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
590 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
592 Updates database nsr with the RO info for the created vld
593 :param ns_update_nsr: dictionary to be filled with the updated info
594 :param db_nsr: content of db_nsr. This is also modified
595 :param nsr_desc_RO: nsr descriptor from RO
596 :return: Nothing, LcmException is raised on errors
599 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
600 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
601 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
603 vld
["vim-id"] = net_RO
.get("vim_net_id")
604 vld
["name"] = net_RO
.get("vim_name")
605 vld
["status"] = net_RO
.get("status")
606 vld
["status-detailed"] = net_RO
.get("error_msg")
607 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
610 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
612 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
614 for db_vnfr
in db_vnfrs
.values():
615 vnfr_update
= {"status": "ERROR"}
616 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
617 if "status" not in vdur
:
618 vdur
["status"] = "ERROR"
619 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
621 vdur
["status-detailed"] = str(error_text
)
622 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
623 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
624 except DbException
as e
:
625 self
.logger
.error("Cannot update vnf. {}".format(e
))
627 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
629 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
630 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
631 :param nsr_desc_RO: nsr descriptor from RO
632 :return: Nothing, LcmException is raised on errors
634 for vnf_index
, db_vnfr
in db_vnfrs
.items():
635 for vnf_RO
in nsr_desc_RO
["vnfs"]:
636 if vnf_RO
["member_vnf_index"] != vnf_index
:
639 if vnf_RO
.get("ip_address"):
640 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
641 elif not db_vnfr
.get("ip-address"):
642 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
643 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
645 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
646 vdur_RO_count_index
= 0
647 if vdur
.get("pdu-type"):
649 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
650 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
652 if vdur
["count-index"] != vdur_RO_count_index
:
653 vdur_RO_count_index
+= 1
655 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
656 if vdur_RO
.get("ip_address"):
657 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
659 vdur
["ip-address"] = None
660 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
661 vdur
["name"] = vdur_RO
.get("vim_name")
662 vdur
["status"] = vdur_RO
.get("status")
663 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
664 for ifacer
in get_iterable(vdur
, "interfaces"):
665 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
666 if ifacer
["name"] == interface_RO
.get("internal_name"):
667 ifacer
["ip-address"] = interface_RO
.get("ip_address")
668 ifacer
["mac-address"] = interface_RO
.get("mac_address")
671 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
673 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
674 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
677 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
678 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
680 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
681 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
682 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
684 vld
["vim-id"] = net_RO
.get("vim_net_id")
685 vld
["name"] = net_RO
.get("vim_name")
686 vld
["status"] = net_RO
.get("status")
687 vld
["status-detailed"] = net_RO
.get("error_msg")
688 vnfr_update
["vld.{}".format(vld_index
)] = vld
691 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
692 vnf_index
, vld
["id"]))
694 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
698 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
700 def _get_ns_config_info(self
, nsr_id
):
702 Generates a mapping between vnf,vdu elements and the N2VC id
703 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
704 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
705 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
706 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
708 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
709 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
711 ns_config_info
= {"osm-config-mapping": mapping
}
712 for vca
in vca_deployed_list
:
713 if not vca
["member-vnf-index"]:
715 if not vca
["vdu_id"]:
716 mapping
[vca
["member-vnf-index"]] = vca
["application"]
718 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
720 return ns_config_info
723 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
):
725 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
726 primitives as verify-ssh-credentials, or config when needed
727 :param desc_primitive_list: information of the descriptor
728 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
729 this element contains a ssh public key
730 :return: The modified list. Can ba an empty list, but always a list
732 if desc_primitive_list
:
733 primitive_list
= desc_primitive_list
.copy()
736 # look for primitive config, and get the position. None if not present
737 config_position
= None
738 for index
, primitive
in enumerate(primitive_list
):
739 if primitive
["name"] == "config":
740 config_position
= index
743 # for NS, add always a config primitive if not present (bug 874)
744 if not vca_deployed
["member-vnf-index"] and config_position
is None:
745 primitive_list
.insert(0, {"name": "config", "parameter": []})
747 # for VNF/VDU add verify-ssh-credentials after config
748 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
749 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
750 return primitive_list
752 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
753 n2vc_key_list
, stage
):
756 RO_descriptor_number
= 0 # number of descriptors created at RO
757 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
758 nslcmop_id
= db_nslcmop
["_id"]
759 start_deploy
= time()
760 ns_params
= db_nslcmop
.get("operationParams")
761 if ns_params
and ns_params
.get("timeout_ns_deploy"):
762 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
764 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
766 # Check for and optionally request placement optimization. Database will be updated if placement activated
767 stage
[2] = "Waiting for Placement."
768 await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
)
772 # get vnfds, instantiate at RO
773 for c_vnf
in nsd
.get("constituent-vnfd", ()):
774 member_vnf_index
= c_vnf
["member-vnf-index"]
775 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
776 vnfd_ref
= vnfd
["id"]
778 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
779 db_nsr_update
["detailed-status"] = " ".join(stage
)
780 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
781 self
._write
_op
_status
(nslcmop_id
, stage
)
783 # self.logger.debug(logging_text + stage[2])
784 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
785 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
786 RO_descriptor_number
+= 1
788 # look position at deployed.RO.vnfd if not present it will be appended at the end
789 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
790 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
793 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
794 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
797 RO_update
= {"member-vnf-index": member_vnf_index
}
798 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
800 RO_update
["id"] = vnfd_list
[0]["uuid"]
801 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
802 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
804 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
805 get("additionalParamsForVnf"), nsr_id
)
806 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
807 RO_update
["id"] = desc
["uuid"]
808 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
809 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
810 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
811 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
816 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
817 db_nsr_update
["detailed-status"] = " ".join(stage
)
818 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
819 self
._write
_op
_status
(nslcmop_id
, stage
)
821 # self.logger.debug(logging_text + stage[2])
822 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
823 RO_descriptor_number
+= 1
824 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
826 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
827 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
828 nsd_ref
, RO_nsd_uuid
))
830 nsd_RO
= deepcopy(nsd
)
831 nsd_RO
["id"] = RO_osm_nsd_id
832 nsd_RO
.pop("_id", None)
833 nsd_RO
.pop("_admin", None)
834 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
835 member_vnf_index
= c_vnf
["member-vnf-index"]
836 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
837 for c_vld
in nsd_RO
.get("vld", ()):
838 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
839 member_vnf_index
= cp
["member-vnf-index-ref"]
840 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
842 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
843 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
844 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
845 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
846 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
849 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
850 db_nsr_update
["detailed-status"] = " ".join(stage
)
851 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
852 self
._write
_op
_status
(nslcmop_id
, stage
)
854 # if present use it unless in error status
855 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
858 stage
[2] = "Looking for existing ns at RO"
859 db_nsr_update
["detailed-status"] = " ".join(stage
)
860 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
861 self
._write
_op
_status
(nslcmop_id
, stage
)
862 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
863 desc
= await self
.RO
.show("ns", RO_nsr_id
)
865 except ROclient
.ROClientException
as e
:
866 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
868 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
870 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
871 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
872 if ns_status
== "ERROR":
873 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
874 self
.logger
.debug(logging_text
+ stage
[2])
875 await self
.RO
.delete("ns", RO_nsr_id
)
876 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
878 stage
[2] = "Checking dependencies"
879 db_nsr_update
["detailed-status"] = " ".join(stage
)
880 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
881 self
._write
_op
_status
(nslcmop_id
, stage
)
882 # self.logger.debug(logging_text + stage[2])
884 # check if VIM is creating and wait look if previous tasks in process
885 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
887 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
888 self
.logger
.debug(logging_text
+ stage
[2])
889 await asyncio
.wait(task_dependency
, timeout
=3600)
890 if ns_params
.get("vnf"):
891 for vnf
in ns_params
["vnf"]:
892 if "vimAccountId" in vnf
:
893 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
896 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
897 self
.logger
.debug(logging_text
+ stage
[2])
898 await asyncio
.wait(task_dependency
, timeout
=3600)
900 stage
[2] = "Checking instantiation parameters."
901 RO_ns_params
= self
.ns_params_2_RO(ns_params
, nsd
, db_vnfds_ref
, n2vc_key_list
)
902 stage
[2] = "Deploying ns at VIM."
903 db_nsr_update
["detailed-status"] = " ".join(stage
)
904 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
905 self
._write
_op
_status
(nslcmop_id
, stage
)
907 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
908 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
909 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
910 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
911 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
913 # wait until NS is ready
914 stage
[2] = "Waiting VIM to deploy ns."
915 db_nsr_update
["detailed-status"] = " ".join(stage
)
916 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
917 self
._write
_op
_status
(nslcmop_id
, stage
)
918 detailed_status_old
= None
919 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
922 while time() <= start_deploy
+ timeout_ns_deploy
:
923 desc
= await self
.RO
.show("ns", RO_nsr_id
)
927 # desc has changed => update db
928 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
931 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
932 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
933 if ns_status
== "ERROR":
934 raise ROclient
.ROClientException(ns_status_info
)
935 elif ns_status
== "BUILD":
936 stage
[2] = "VIM: ({})".format(ns_status_info
)
937 elif ns_status
== "ACTIVE":
938 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
940 self
.ns_update_vnfr(db_vnfrs
, desc
)
942 except LcmExceptionNoMgmtIP
:
945 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
946 if stage
[2] != detailed_status_old
:
947 detailed_status_old
= stage
[2]
948 db_nsr_update
["detailed-status"] = " ".join(stage
)
949 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
950 self
._write
_op
_status
(nslcmop_id
, stage
)
951 await asyncio
.sleep(5, loop
=self
.loop
)
952 else: # timeout_ns_deploy
953 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
956 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
958 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
959 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
960 stage
[2] = "Deployed at VIM"
961 db_nsr_update
["detailed-status"] = " ".join(stage
)
962 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
963 self
._write
_op
_status
(nslcmop_id
, stage
)
964 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
965 # self.logger.debug(logging_text + "Deployed at VIM")
966 except (ROclient
.ROClientException
, LcmException
, DbException
) as e
:
967 stage
[2] = "ERROR deploying at VIM"
968 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
971 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
973 Wait for ip addres at RO, and optionally, insert public key in virtual machine
974 :param logging_text: prefix use for logging
979 :param pub_key: public ssh key to inject, None to skip
980 :param user: user to apply the public ssh key
984 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
994 if ro_retries
>= 360: # 1 hour
995 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
997 await asyncio
.sleep(10, loop
=self
.loop
)
1000 if not target_vdu_id
:
1001 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1003 if not vdu_id
: # for the VNF case
1004 if db_vnfr
.get("status") == "ERROR":
1005 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1006 ip_address
= db_vnfr
.get("ip-address")
1009 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1011 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1012 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1014 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1015 vdur
= db_vnfr
["vdur"][0]
1017 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1020 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1021 ip_address
= vdur
.get("ip-address")
1024 target_vdu_id
= vdur
["vdu-id-ref"]
1025 elif vdur
.get("status") == "ERROR":
1026 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1028 if not target_vdu_id
:
1031 # inject public key into machine
1032 if pub_key
and user
:
1033 # wait until NS is deployed at RO
1035 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1036 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1040 # self.logger.debug(logging_text + "Inserting RO key")
1041 if vdur
.get("pdu-type"):
1042 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1045 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1046 result_dict
= await self
.RO
.create_action(
1048 item_id_name
=ro_nsr_id
,
1049 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1051 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1052 if not result_dict
or not isinstance(result_dict
, dict):
1053 raise LcmException("Unknown response from RO when injecting key")
1054 for result
in result_dict
.values():
1055 if result
.get("vim_result") == 200:
1058 raise ROclient
.ROClientException("error injecting key: {}".format(
1059 result
.get("description")))
1061 except ROclient
.ROClientException
as e
:
1063 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1067 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1073 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1075 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1077 my_vca
= vca_deployed_list
[vca_index
]
1078 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1079 # vdu or kdu: no dependencies
1083 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1084 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1085 configuration_status_list
= db_nsr
["configurationStatus"]
1086 for index
, vca_deployed
in enumerate(configuration_status_list
):
1087 if index
== vca_index
:
1090 if not my_vca
.get("member-vnf-index") or \
1091 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1092 internal_status
= configuration_status_list
[index
].get("status")
1093 if internal_status
== 'READY':
1095 elif internal_status
== 'BROKEN':
1096 raise LcmException("Configuration aborted because dependent charm/s has failed")
1100 # no dependencies, return
1102 await asyncio
.sleep(10)
1105 raise LcmException("Configuration aborted because dependent charm/s timeout")
1107 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1108 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
):
1109 nsr_id
= db_nsr
["_id"]
1110 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1111 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1112 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1114 'collection': 'nsrs',
1115 'filter': {'_id': nsr_id
},
1116 'path': db_update_entry
1122 element_under_configuration
= nsr_id
1126 vnfr_id
= db_vnfr
["_id"]
1128 namespace
= "{nsi}.{ns}".format(
1129 nsi
=nsi_id
if nsi_id
else "",
1133 element_type
= 'VNF'
1134 element_under_configuration
= vnfr_id
1135 namespace
+= ".{}".format(vnfr_id
)
1137 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1138 element_type
= 'VDU'
1139 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1141 namespace
+= ".{}".format(kdu_name
)
1142 element_type
= 'KDU'
1143 element_under_configuration
= kdu_name
1146 self
.fs
.sync() # Sync from FSMongo
1147 artifact_path
= "{}/{}/charms/{}".format(
1148 base_folder
["folder"],
1149 base_folder
["pkg-dir"],
1150 config_descriptor
["juju"]["charm"]
1153 is_proxy_charm
= deep_get(config_descriptor
, ('juju', 'charm')) is not None
1154 if deep_get(config_descriptor
, ('juju', 'proxy')) is False:
1155 is_proxy_charm
= False
1157 # n2vc_redesign STEP 3.1
1159 # find old ee_id if exists
1160 ee_id
= vca_deployed
.get("ee_id")
1162 # create or register execution environment in VCA
1165 self
._write
_configuration
_status
(
1167 vca_index
=vca_index
,
1169 element_under_configuration
=element_under_configuration
,
1170 element_type
=element_type
1173 step
= "create execution environment"
1174 self
.logger
.debug(logging_text
+ step
)
1175 ee_id
, credentials
= await self
.n2vc
.create_execution_environment(namespace
=namespace
,
1180 step
= "Waiting to VM being up and getting IP address"
1181 self
.logger
.debug(logging_text
+ step
)
1182 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1183 user
=None, pub_key
=None)
1184 credentials
= {"hostname": rw_mgmt_ip
}
1186 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1187 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1188 # merged. Meanwhile let's get username from initial-config-primitive
1189 if not username
and config_descriptor
.get("initial-config-primitive"):
1190 for config_primitive
in config_descriptor
["initial-config-primitive"]:
1191 for param
in config_primitive
.get("parameter", ()):
1192 if param
["name"] == "ssh-username":
1193 username
= param
["value"]
1196 raise LcmException("Cannot determine the username neither with 'initial-config-promitive' nor with "
1197 "'config-access.ssh-access.default-user'")
1198 credentials
["username"] = username
1199 # n2vc_redesign STEP 3.2
1201 self
._write
_configuration
_status
(
1203 vca_index
=vca_index
,
1204 status
='REGISTERING',
1205 element_under_configuration
=element_under_configuration
,
1206 element_type
=element_type
1209 step
= "register execution environment {}".format(credentials
)
1210 self
.logger
.debug(logging_text
+ step
)
1211 ee_id
= await self
.n2vc
.register_execution_environment(credentials
=credentials
, namespace
=namespace
,
1214 # for compatibility with MON/POL modules, the need model and application name at database
1215 # TODO ask to N2VC instead of assuming the format "model_name.application_name"
1216 ee_id_parts
= ee_id
.split('.')
1217 model_name
= ee_id_parts
[0]
1218 application_name
= ee_id_parts
[1]
1219 db_nsr_update
= {db_update_entry
+ "model": model_name
,
1220 db_update_entry
+ "application": application_name
,
1221 db_update_entry
+ "ee_id": ee_id
}
1223 # n2vc_redesign STEP 3.3
1225 step
= "Install configuration Software"
1227 self
._write
_configuration
_status
(
1229 vca_index
=vca_index
,
1230 status
='INSTALLING SW',
1231 element_under_configuration
=element_under_configuration
,
1232 element_type
=element_type
,
1233 other_update
=db_nsr_update
1236 # TODO check if already done
1237 self
.logger
.debug(logging_text
+ step
)
1239 if not is_proxy_charm
:
1240 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1241 if initial_config_primitive_list
:
1242 for primitive
in initial_config_primitive_list
:
1243 if primitive
["name"] == "config":
1244 config
= self
._map
_primitive
_params
(
1250 await self
.n2vc
.install_configuration_sw(
1252 artifact_path
=artifact_path
,
1257 # write in db flag of configuration_sw already installed
1258 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1260 # add relations for this VCA (wait for other peers related with this VCA)
1261 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
, vca_index
=vca_index
)
1263 # if SSH access is required, then get execution environment SSH public
1264 if is_proxy_charm
: # if native charm we have waited already to VM be UP
1267 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1268 # Needed to inject a ssh key
1269 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1270 step
= "Install configuration Software, getting public ssh key"
1271 pub_key
= await self
.n2vc
.get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1273 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1275 step
= "Waiting to VM being up and getting IP address"
1276 self
.logger
.debug(logging_text
+ step
)
1278 # n2vc_redesign STEP 5.1
1279 # wait for RO (ip-address) Insert pub_key into VM
1281 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1282 user
=user
, pub_key
=pub_key
)
1284 rw_mgmt_ip
= None # This is for a NS configuration
1286 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1288 # store rw_mgmt_ip in deploy params for later replacement
1289 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1291 # n2vc_redesign STEP 6 Execute initial config primitive
1292 step
= 'execute initial config primitive'
1293 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1295 # sort initial config primitives by 'seq'
1296 if initial_config_primitive_list
:
1298 initial_config_primitive_list
.sort(key
=lambda val
: int(val
['seq']))
1299 except Exception as e
:
1300 self
.logger
.error(logging_text
+ step
+ ": " + str(e
))
1302 self
.logger
.debug(logging_text
+ step
+ ": No initial-config-primitive")
1304 # add config if not present for NS charm
1305 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1308 # wait for dependent primitives execution (NS -> VNF -> VDU)
1309 if initial_config_primitive_list
:
1310 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1312 # stage, in function of element type: vdu, kdu, vnf or ns
1313 my_vca
= vca_deployed_list
[vca_index
]
1314 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1316 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1317 elif my_vca
.get("member-vnf-index"):
1319 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1322 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1324 self
._write
_configuration
_status
(
1326 vca_index
=vca_index
,
1327 status
='EXECUTING PRIMITIVE'
1330 self
._write
_op
_status
(
1335 check_if_terminated_needed
= True
1336 for initial_config_primitive
in initial_config_primitive_list
:
1337 # adding information on the vca_deployed if it is a NS execution environment
1338 if not vca_deployed
["member-vnf-index"]:
1339 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1340 # TODO check if already done
1341 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1343 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1344 self
.logger
.debug(logging_text
+ step
)
1345 await self
.n2vc
.exec_primitive(
1347 primitive_name
=initial_config_primitive
["name"],
1348 params_dict
=primitive_params_
,
1351 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1352 if check_if_terminated_needed
:
1353 if config_descriptor
.get('terminate-config-primitive'):
1354 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1355 check_if_terminated_needed
= False
1357 # TODO register in database that primitive is done
1359 step
= "instantiated at VCA"
1360 self
.logger
.debug(logging_text
+ step
)
1362 self
._write
_configuration
_status
(
1364 vca_index
=vca_index
,
1368 except Exception as e
: # TODO not use Exception but N2VC exception
1369 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1370 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1371 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1372 self
._write
_configuration
_status
(
1374 vca_index
=vca_index
,
1377 raise LcmException("{} {}".format(step
, e
)) from e
1379 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1380 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1382 Update db_nsr fields.
1385 :param current_operation:
1386 :param current_operation_id:
1387 :param error_description:
1388 :param error_detail:
1389 :param other_update: Other required changes at database if provided, will be cleared
1393 db_dict
= other_update
or {}
1394 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1395 db_dict
["_admin.current-operation"] = current_operation_id
1396 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1397 db_dict
["currentOperation"] = current_operation
1398 db_dict
["currentOperationID"] = current_operation_id
1399 db_dict
["errorDescription"] = error_description
1400 db_dict
["errorDetail"] = error_detail
1403 db_dict
["nsState"] = ns_state
1404 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1405 except DbException
as e
:
1406 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1408 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1409 operation_state
: str = None, other_update
: dict = None):
1411 db_dict
= other_update
or {}
1412 db_dict
['queuePosition'] = queuePosition
1413 if isinstance(stage
, list):
1414 db_dict
['stage'] = stage
[0]
1415 db_dict
['detailed-status'] = " ".join(stage
)
1416 elif stage
is not None:
1417 db_dict
['stage'] = str(stage
)
1419 if error_message
is not None:
1420 db_dict
['errorMessage'] = error_message
1421 if operation_state
is not None:
1422 db_dict
['operationState'] = operation_state
1423 db_dict
["statusEnteredTime"] = time()
1424 self
.update_db_2("nslcmops", op_id
, db_dict
)
1425 except DbException
as e
:
1426 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1428 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1430 nsr_id
= db_nsr
["_id"]
1431 # configurationStatus
1432 config_status
= db_nsr
.get('configurationStatus')
1434 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1435 enumerate(config_status
) if v
}
1437 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1439 except DbException
as e
:
1440 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1442 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1443 element_under_configuration
: str = None, element_type
: str = None,
1444 other_update
: dict = None):
1446 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1447 # .format(vca_index, status))
1450 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1451 db_dict
= other_update
or {}
1453 db_dict
[db_path
+ 'status'] = status
1454 if element_under_configuration
:
1455 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1457 db_dict
[db_path
+ 'elementType'] = element_type
1458 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1459 except DbException
as e
:
1460 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1461 .format(status
, nsr_id
, vca_index
, e
))
1463 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1465 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1466 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1467 Database is used because the result can be obtained from a different LCM worker in case of HA.
1468 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1469 :param db_nslcmop: database content of nslcmop
1470 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1471 :return: None. Modifies database vnfrs and parameter db_vnfr with the computed 'vim-account-id'
1473 nslcmop_id
= db_nslcmop
['_id']
1474 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1475 if placement_engine
== "PLA":
1476 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1477 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1478 db_poll_interval
= 5
1479 wait
= db_poll_interval
* 10
1481 while not pla_result
and wait
>= 0:
1482 await asyncio
.sleep(db_poll_interval
)
1483 wait
-= db_poll_interval
1484 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1485 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1488 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1490 for pla_vnf
in pla_result
['vnf']:
1491 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1492 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1494 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1496 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1499 def update_nsrs_with_pla_result(self
, params
):
1501 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1502 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1503 except Exception as e
:
1504 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1506 async def instantiate(self
, nsr_id
, nslcmop_id
):
1509 :param nsr_id: ns instance to deploy
1510 :param nslcmop_id: operation to run
1514 # Try to lock HA task here
1515 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1516 if not task_is_locked_by_me
:
1517 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1520 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1521 self
.logger
.debug(logging_text
+ "Enter")
1523 # get all needed from database
1525 # database nsrs record
1528 # database nslcmops record
1531 # update operation on nsrs
1533 # update operation on nslcmops
1534 db_nslcmop_update
= {}
1536 nslcmop_operation_state
= None
1537 db_vnfrs
= {} # vnf's info indexed by member-index
1539 tasks_dict_info
= {} # from task to info text
1542 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1543 # ^ stage, step, VIM progress
1545 # wait for any previous tasks in process
1546 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1548 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1549 stage
[1] = "Reading from database,"
1550 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1551 db_nsr_update
["detailed-status"] = "creating"
1552 db_nsr_update
["operational-status"] = "init"
1553 self
._write
_ns
_status
(
1555 ns_state
="BUILDING",
1556 current_operation
="INSTANTIATING",
1557 current_operation_id
=nslcmop_id
,
1558 other_update
=db_nsr_update
1560 self
._write
_op
_status
(
1566 # read from db: operation
1567 stage
[1] = "Getting nslcmop={} from db".format(nslcmop_id
)
1568 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1569 ns_params
= db_nslcmop
.get("operationParams")
1570 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1571 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1573 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1576 stage
[1] = "Getting nsr={} from db".format(nsr_id
)
1577 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1578 # nsd is replicated into ns (no db read)
1580 # nsr_name = db_nsr["name"] # TODO short-name??
1582 # read from db: vnf's of this ns
1583 stage
[1] = "Getting vnfrs from db"
1584 self
.logger
.debug(logging_text
+ stage
[1])
1585 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1587 # read from db: vnfd's for every vnf
1588 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1589 db_vnfds
= {} # every vnfd data indexed by vnf id
1590 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1592 # for each vnf in ns, read vnfd
1593 for vnfr
in db_vnfrs_list
:
1594 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1595 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1596 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1597 # if we haven't this vnfd, read it from db
1598 if vnfd_id
not in db_vnfds
:
1600 stage
[1] = "Getting vnfd={} id='{}' from db".format(vnfd_id
, vnfd_ref
)
1601 self
.logger
.debug(logging_text
+ stage
[1])
1602 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1605 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1606 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1607 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1609 # Get or generates the _admin.deployed.VCA list
1610 vca_deployed_list
= None
1611 if db_nsr
["_admin"].get("deployed"):
1612 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1613 if vca_deployed_list
is None:
1614 vca_deployed_list
= []
1615 configuration_status_list
= []
1616 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1617 db_nsr_update
["configurationStatus"] = configuration_status_list
1618 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1619 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1620 elif isinstance(vca_deployed_list
, dict):
1621 # maintain backward compatibility. Change a dict to list at database
1622 vca_deployed_list
= list(vca_deployed_list
.values())
1623 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1624 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1626 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1627 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1628 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1630 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1631 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1632 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1634 # n2vc_redesign STEP 2 Deploy Network Scenario
1635 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1636 self
._write
_op
_status
(
1641 stage
[1] = "Deploying KDUs,"
1642 # self.logger.debug(logging_text + "Before deploy_kdus")
1643 # Call to deploy_kdus in case exists the "vdu:kdu" param
1644 await self
.deploy_kdus(
1645 logging_text
=logging_text
,
1647 nslcmop_id
=nslcmop_id
,
1650 task_instantiation_info
=tasks_dict_info
,
1653 stage
[1] = "Getting VCA public key."
1654 # n2vc_redesign STEP 1 Get VCA public ssh-key
1655 # feature 1429. Add n2vc public key to needed VMs
1656 n2vc_key
= self
.n2vc
.get_public_key()
1657 n2vc_key_list
= [n2vc_key
]
1658 if self
.vca_config
.get("public_key"):
1659 n2vc_key_list
.append(self
.vca_config
["public_key"])
1661 stage
[1] = "Deploying NS at VIM."
1662 task_ro
= asyncio
.ensure_future(
1663 self
.instantiate_RO(
1664 logging_text
=logging_text
,
1668 db_nslcmop
=db_nslcmop
,
1670 db_vnfds_ref
=db_vnfds_ref
,
1671 n2vc_key_list
=n2vc_key_list
,
1675 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
1676 tasks_dict_info
[task_ro
] = "Deploying at VIM"
1678 # n2vc_redesign STEP 3 to 6 Deploy N2VC
1679 stage
[1] = "Deploying Execution Environments."
1680 self
.logger
.debug(logging_text
+ stage
[1])
1682 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
1683 # get_iterable() returns a value from a dict or empty tuple if key does not exist
1684 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
1685 vnfd_id
= c_vnf
["vnfd-id-ref"]
1686 vnfd
= db_vnfds_ref
[vnfd_id
]
1687 member_vnf_index
= str(c_vnf
["member-vnf-index"])
1688 db_vnfr
= db_vnfrs
[member_vnf_index
]
1689 base_folder
= vnfd
["_admin"]["storage"]
1695 # Get additional parameters
1697 if db_vnfr
.get("additionalParamsForVnf"):
1698 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
1700 descriptor_config
= vnfd
.get("vnf-configuration")
1701 if descriptor_config
and descriptor_config
.get("juju"):
1703 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
1706 nslcmop_id
=nslcmop_id
,
1712 member_vnf_index
=member_vnf_index
,
1713 vdu_index
=vdu_index
,
1715 deploy_params
=deploy_params
,
1716 descriptor_config
=descriptor_config
,
1717 base_folder
=base_folder
,
1718 task_instantiation_info
=tasks_dict_info
,
1722 # Deploy charms for each VDU that supports one.
1723 for vdud
in get_iterable(vnfd
, 'vdu'):
1725 descriptor_config
= vdud
.get('vdu-configuration')
1726 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
1727 if vdur
.get("additionalParams"):
1728 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
1730 deploy_params_vdu
= deploy_params
1731 if descriptor_config
and descriptor_config
.get("juju"):
1732 # look for vdu index in the db_vnfr["vdu"] section
1733 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1734 # if vdur["vdu-id-ref"] == vdu_id:
1737 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1738 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1739 # vdu_name = vdur.get("name")
1742 for vdu_index
in range(int(vdud
.get("count", 1))):
1743 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
1745 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
1746 member_vnf_index
, vdu_id
, vdu_index
),
1749 nslcmop_id
=nslcmop_id
,
1755 member_vnf_index
=member_vnf_index
,
1756 vdu_index
=vdu_index
,
1758 deploy_params
=deploy_params_vdu
,
1759 descriptor_config
=descriptor_config
,
1760 base_folder
=base_folder
,
1761 task_instantiation_info
=tasks_dict_info
,
1764 for kdud
in get_iterable(vnfd
, 'kdu'):
1765 kdu_name
= kdud
["name"]
1766 descriptor_config
= kdud
.get('kdu-configuration')
1767 if descriptor_config
and descriptor_config
.get("juju"):
1771 # look for vdu index in the db_vnfr["vdu"] section
1772 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1773 # if vdur["vdu-id-ref"] == vdu_id:
1776 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1777 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1778 # vdu_name = vdur.get("name")
1782 logging_text
=logging_text
,
1785 nslcmop_id
=nslcmop_id
,
1791 member_vnf_index
=member_vnf_index
,
1792 vdu_index
=vdu_index
,
1794 deploy_params
=deploy_params
,
1795 descriptor_config
=descriptor_config
,
1796 base_folder
=base_folder
,
1797 task_instantiation_info
=tasks_dict_info
,
1801 # Check if this NS has a charm configuration
1802 descriptor_config
= nsd
.get("ns-configuration")
1803 if descriptor_config
and descriptor_config
.get("juju"):
1806 member_vnf_index
= None
1812 # Get additional parameters
1814 if db_nsr
.get("additionalParamsForNs"):
1815 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
1816 base_folder
= nsd
["_admin"]["storage"]
1818 logging_text
=logging_text
,
1821 nslcmop_id
=nslcmop_id
,
1827 member_vnf_index
=member_vnf_index
,
1828 vdu_index
=vdu_index
,
1830 deploy_params
=deploy_params
,
1831 descriptor_config
=descriptor_config
,
1832 base_folder
=base_folder
,
1833 task_instantiation_info
=tasks_dict_info
,
1837 # rest of staff will be done at finally
1839 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
1840 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
1842 except asyncio
.CancelledError
:
1843 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
1844 exc
= "Operation was cancelled"
1845 except Exception as e
:
1846 exc
= traceback
.format_exc()
1847 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
1850 error_list
.append(str(exc
))
1852 # wait for pending tasks
1854 stage
[1] = "Waiting for instantiate pending tasks."
1855 self
.logger
.debug(logging_text
+ stage
[1])
1856 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
1857 stage
, nslcmop_id
, nsr_id
=nsr_id
)
1858 stage
[1] = stage
[2] = ""
1859 except asyncio
.CancelledError
:
1860 error_list
.append("Cancelled")
1861 # TODO cancel all tasks
1862 except Exception as exc
:
1863 error_list
.append(str(exc
))
1865 # update operation-status
1866 db_nsr_update
["operational-status"] = "running"
1867 # let's begin with VCA 'configured' status (later we can change it)
1868 db_nsr_update
["config-status"] = "configured"
1869 for task
, task_name
in tasks_dict_info
.items():
1870 if not task
.done() or task
.cancelled() or task
.exception():
1871 if task_name
.startswith(self
.task_name_deploy_vca
):
1872 # A N2VC task is pending
1873 db_nsr_update
["config-status"] = "failed"
1875 # RO or KDU task is pending
1876 db_nsr_update
["operational-status"] = "failed"
1878 # update status at database
1880 error_detail
= ". ".join(error_list
)
1881 self
.logger
.error(logging_text
+ error_detail
)
1882 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
1883 error_description_nsr
= 'Operation: INSTANTIATING.{}, Stage {}'.format(nslcmop_id
, stage
[0])
1885 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
1886 db_nslcmop_update
["detailed-status"] = error_detail
1887 nslcmop_operation_state
= "FAILED"
1891 error_description_nsr
= error_description_nslcmop
= None
1893 db_nsr_update
["detailed-status"] = "Done"
1894 db_nslcmop_update
["detailed-status"] = "Done"
1895 nslcmop_operation_state
= "COMPLETED"
1898 self
._write
_ns
_status
(
1901 current_operation
="IDLE",
1902 current_operation_id
=None,
1903 error_description
=error_description_nsr
,
1904 error_detail
=error_detail
,
1905 other_update
=db_nsr_update
1908 self
._write
_op
_status
(
1911 error_message
=error_description_nslcmop
,
1912 operation_state
=nslcmop_operation_state
,
1913 other_update
=db_nslcmop_update
,
1916 if nslcmop_operation_state
:
1918 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
1919 "operationState": nslcmop_operation_state
},
1921 except Exception as e
:
1922 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
1924 self
.logger
.debug(logging_text
+ "Exit")
1925 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
1927 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int, timeout
: int = 3600) -> bool:
1930 # 1. find all relations for this VCA
1931 # 2. wait for other peers related
1936 # STEP 1: find all relations for this VCA
1939 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1942 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
1944 # read all ns-configuration relations
1945 ns_relations
= list()
1946 db_ns_relations
= deep_get(db_nsr
, ('nsd', 'ns-configuration', 'relation'))
1948 for r
in db_ns_relations
:
1949 # check if this VCA is in the relation
1950 if my_vca
.get('member-vnf-index') in\
1951 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1952 ns_relations
.append(r
)
1954 # read all vnf-configuration relations
1955 vnf_relations
= list()
1956 db_vnfd_list
= db_nsr
.get('vnfd-id')
1958 for vnfd
in db_vnfd_list
:
1959 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
1960 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
1961 if db_vnf_relations
:
1962 for r
in db_vnf_relations
:
1963 # check if this VCA is in the relation
1964 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1965 vnf_relations
.append(r
)
1967 # if no relations, terminate
1968 if not ns_relations
and not vnf_relations
:
1969 self
.logger
.debug(logging_text
+ ' No relations')
1972 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
1979 if now
- start
>= timeout
:
1980 self
.logger
.error(logging_text
+ ' : timeout adding relations')
1983 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
1984 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1986 # for each defined NS relation, find the VCA's related
1987 for r
in ns_relations
:
1988 from_vca_ee_id
= None
1990 from_vca_endpoint
= None
1991 to_vca_endpoint
= None
1992 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
1993 for vca
in vca_list
:
1994 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
1995 and vca
.get('config_sw_installed'):
1996 from_vca_ee_id
= vca
.get('ee_id')
1997 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
1998 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
1999 and vca
.get('config_sw_installed'):
2000 to_vca_ee_id
= vca
.get('ee_id')
2001 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2002 if from_vca_ee_id
and to_vca_ee_id
:
2004 await self
.n2vc
.add_relation(
2005 ee_id_1
=from_vca_ee_id
,
2006 ee_id_2
=to_vca_ee_id
,
2007 endpoint_1
=from_vca_endpoint
,
2008 endpoint_2
=to_vca_endpoint
)
2009 # remove entry from relations list
2010 ns_relations
.remove(r
)
2012 # check failed peers
2014 vca_status_list
= db_nsr
.get('configurationStatus')
2016 for i
in range(len(vca_list
)):
2018 vca_status
= vca_status_list
[i
]
2019 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2020 if vca_status
.get('status') == 'BROKEN':
2021 # peer broken: remove relation from list
2022 ns_relations
.remove(r
)
2023 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2024 if vca_status
.get('status') == 'BROKEN':
2025 # peer broken: remove relation from list
2026 ns_relations
.remove(r
)
2031 # for each defined VNF relation, find the VCA's related
2032 for r
in vnf_relations
:
2033 from_vca_ee_id
= None
2035 from_vca_endpoint
= None
2036 to_vca_endpoint
= None
2037 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2038 for vca
in vca_list
:
2039 if vca
.get('vdu_id') == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2040 from_vca_ee_id
= vca
.get('ee_id')
2041 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2042 if vca
.get('vdu_id') == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2043 to_vca_ee_id
= vca
.get('ee_id')
2044 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2045 if from_vca_ee_id
and to_vca_ee_id
:
2047 await self
.n2vc
.add_relation(
2048 ee_id_1
=from_vca_ee_id
,
2049 ee_id_2
=to_vca_ee_id
,
2050 endpoint_1
=from_vca_endpoint
,
2051 endpoint_2
=to_vca_endpoint
)
2052 # remove entry from relations list
2053 vnf_relations
.remove(r
)
2055 # check failed peers
2057 vca_status_list
= db_nsr
.get('configurationStatus')
2059 for i
in range(len(vca_list
)):
2061 vca_status
= vca_status_list
[i
]
2062 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2063 if vca_status
.get('status') == 'BROKEN':
2064 # peer broken: remove relation from list
2065 ns_relations
.remove(r
)
2066 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2067 if vca_status
.get('status') == 'BROKEN':
2068 # peer broken: remove relation from list
2069 ns_relations
.remove(r
)
2075 await asyncio
.sleep(5.0)
2077 if not ns_relations
and not vnf_relations
:
2078 self
.logger
.debug('Relations added')
2083 except Exception as e
:
2084 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2087 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2088 # Launch kdus if present in the descriptor
2090 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2092 def _get_cluster_id(cluster_id
, cluster_type
):
2093 nonlocal k8scluster_id_2_uuic
2094 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2095 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2097 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2098 if not db_k8scluster
:
2099 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2100 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2102 raise LcmException("K8s cluster '{}' has not been initilized for '{}'".format(cluster_id
, cluster_type
))
2103 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2106 logging_text
+= "Deploy kdus: "
2109 db_nsr_update
= {"_admin.deployed.K8s": []}
2110 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2113 updated_cluster_list
= []
2115 for vnfr_data
in db_vnfrs
.values():
2116 for kdur
in get_iterable(vnfr_data
, "kdur"):
2117 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2118 vnfd_id
= vnfr_data
.get('vnfd-id')
2119 if kdur
.get("helm-chart"):
2120 kdumodel
= kdur
["helm-chart"]
2121 k8sclustertype
= "helm-chart"
2122 elif kdur
.get("juju-bundle"):
2123 kdumodel
= kdur
["juju-bundle"]
2124 k8sclustertype
= "juju-bundle"
2126 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2127 "juju-bundle. Maybe an old NBI version is running".
2128 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2129 # check if kdumodel is a file and exists
2131 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2132 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2133 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2134 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["'pkg-dir"], k8sclustertype
,
2136 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2137 kdumodel
= self
.fs
.path
+ filename
2138 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2140 except Exception: # it is not a file
2143 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2144 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2145 cluster_uuid
= _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2147 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2148 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2149 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2150 if del_repo_list
or added_repo_dict
:
2151 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2152 updated
= {'_admin.helm_charts_added.' +
2153 item
: name
for item
, name
in added_repo_dict
.items()}
2154 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2155 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2157 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2158 updated_cluster_list
.append(cluster_uuid
)
2160 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2161 kdur
["kdu-name"], k8s_cluster_id
)
2163 k8s_instace_info
= {"kdu-instance": None,
2164 "k8scluster-uuid": cluster_uuid
,
2165 "k8scluster-type": k8sclustertype
,
2166 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2167 "kdu-name": kdur
["kdu-name"],
2168 "kdu-model": kdumodel
}
2169 db_nsr_update
["_admin.deployed.K8s.{}".format(index
)] = k8s_instace_info
2170 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2172 db_dict
= {"collection": "nsrs",
2173 "filter": {"_id": nsr_id
},
2174 "path": "_admin.deployed.K8s.{}".format(index
)}
2176 task
= asyncio
.ensure_future(
2177 self
.k8scluster_map
[k8sclustertype
].install(cluster_uuid
=cluster_uuid
, kdu_model
=kdumodel
,
2178 atomic
=True, params
=desc_params
,
2179 db_dict
=db_dict
, timeout
=600,
2180 kdu_name
=kdur
["kdu-name"]))
2182 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2183 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2187 except (LcmException
, asyncio
.CancelledError
):
2189 except Exception as e
:
2190 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2191 if isinstance(e
, (N2VCException
, DbException
)):
2192 self
.logger
.error(logging_text
+ msg
)
2194 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2195 raise LcmException(msg
)
2198 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2200 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2201 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2202 base_folder
, task_instantiation_info
, stage
):
2203 # launch instantiate_N2VC in a asyncio task and register task object
2204 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2205 # if not found, create one entry and update database
2207 # fill db_nsr._admin.deployed.VCA.<index>
2209 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2210 if not vca_deployed
:
2212 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2213 vca_deployed
.get("vdu_id") == vdu_id
and \
2214 vca_deployed
.get("kdu_name") == kdu_name
and \
2215 vca_deployed
.get("vdu_count_index", 0) == vdu_index
:
2218 # not found, create one.
2220 "member-vnf-index": member_vnf_index
,
2222 "kdu_name": kdu_name
,
2223 "vdu_count_index": vdu_index
,
2224 "operational-status": "init", # TODO revise
2225 "detailed-status": "", # TODO revise
2226 "step": "initial-deploy", # TODO revise
2228 "vdu_name": vdu_name
,
2232 # create VCA and configurationStatus in db
2234 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2235 "configurationStatus.{}".format(vca_index
): dict()
2237 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2239 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2242 task_n2vc
= asyncio
.ensure_future(
2243 self
.instantiate_N2VC(
2244 logging_text
=logging_text
,
2245 vca_index
=vca_index
,
2251 vdu_index
=vdu_index
,
2252 deploy_params
=deploy_params
,
2253 config_descriptor
=descriptor_config
,
2254 base_folder
=base_folder
,
2255 nslcmop_id
=nslcmop_id
,
2259 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2260 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2261 member_vnf_index
or "", vdu_id
or "")
2263 # Check if this VNFD has a configured terminate action
2264 def _has_terminate_config_primitive(self
, vnfd
):
2265 vnf_config
= vnfd
.get("vnf-configuration")
2266 if vnf_config
and vnf_config
.get("terminate-config-primitive"):
2272 def _get_terminate_config_primitive_seq_list(vnfd
):
2273 """ Get a numerically sorted list of the sequences for this VNFD's terminate action """
2274 # No need to check for existing primitive twice, already done before
2275 vnf_config
= vnfd
.get("vnf-configuration")
2276 seq_list
= vnf_config
.get("terminate-config-primitive")
2277 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2278 seq_list_sorted
= sorted(seq_list
, key
=lambda x
: int(x
['seq']))
2279 return seq_list_sorted
2282 def _create_nslcmop(nsr_id
, operation
, params
):
2284 Creates a ns-lcm-opp content to be stored at database.
2285 :param nsr_id: internal id of the instance
2286 :param operation: instantiate, terminate, scale, action, ...
2287 :param params: user parameters for the operation
2288 :return: dictionary following SOL005 format
2290 # Raise exception if invalid arguments
2291 if not (nsr_id
and operation
and params
):
2293 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2299 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2300 "operationState": "PROCESSING",
2301 "statusEnteredTime": now
,
2302 "nsInstanceId": nsr_id
,
2303 "lcmOperationType": operation
,
2305 "isAutomaticInvocation": False,
2306 "operationParams": params
,
2307 "isCancelPending": False,
2309 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2310 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2315 def _format_additional_params(self
, params
):
2316 params
= params
or {}
2317 for key
, value
in params
.items():
2318 if str(value
).startswith("!!yaml "):
2319 params
[key
] = yaml
.safe_load(value
[7:])
2322 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2323 primitive
= seq
.get('name')
2324 primitive_params
= {}
2326 "member_vnf_index": vnf_index
,
2327 "primitive": primitive
,
2328 "primitive_params": primitive_params
,
2331 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2335 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2336 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2337 if op
.get('operationState') == 'COMPLETED':
2338 # b. Skip sub-operation
2339 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2340 return self
.SUBOPERATION_STATUS_SKIP
2342 # c. Reintent executing sub-operation
2343 # The sub-operation exists, and operationState != 'COMPLETED'
2344 # Update operationState = 'PROCESSING' to indicate a reintent.
2345 operationState
= 'PROCESSING'
2346 detailed_status
= 'In progress'
2347 self
._update
_suboperation
_status
(
2348 db_nslcmop
, op_index
, operationState
, detailed_status
)
2349 # Return the sub-operation index
2350 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2351 # with arguments extracted from the sub-operation
2354 # Find a sub-operation where all keys in a matching dictionary must match
2355 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2356 def _find_suboperation(self
, db_nslcmop
, match
):
2357 if (db_nslcmop
and match
):
2358 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2359 for i
, op
in enumerate(op_list
):
2360 if all(op
.get(k
) == match
[k
] for k
in match
):
2362 return self
.SUBOPERATION_STATUS_NOT_FOUND
2364 # Update status for a sub-operation given its index
2365 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2366 # Update DB for HA tasks
2367 q_filter
= {'_id': db_nslcmop
['_id']}
2368 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2369 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2370 self
.db
.set_one("nslcmops",
2372 update_dict
=update_dict
,
2373 fail_on_empty
=False)
2375 # Add sub-operation, return the index of the added sub-operation
2376 # Optionally, set operationState, detailed-status, and operationType
2377 # Status and type are currently set for 'scale' sub-operations:
2378 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2379 # 'detailed-status' : status message
2380 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2381 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2382 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2383 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2384 RO_nsr_id
=None, RO_scaling_info
=None):
2386 return self
.SUBOPERATION_STATUS_NOT_FOUND
2387 # Get the "_admin.operations" list, if it exists
2388 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2389 op_list
= db_nslcmop_admin
.get('operations')
2390 # Create or append to the "_admin.operations" list
2391 new_op
= {'member_vnf_index': vnf_index
,
2393 'vdu_count_index': vdu_count_index
,
2394 'primitive': primitive
,
2395 'primitive_params': mapped_primitive_params
}
2397 new_op
['operationState'] = operationState
2399 new_op
['detailed-status'] = detailed_status
2401 new_op
['lcmOperationType'] = operationType
2403 new_op
['RO_nsr_id'] = RO_nsr_id
2405 new_op
['RO_scaling_info'] = RO_scaling_info
2407 # No existing operations, create key 'operations' with current operation as first list element
2408 db_nslcmop_admin
.update({'operations': [new_op
]})
2409 op_list
= db_nslcmop_admin
.get('operations')
2411 # Existing operations, append operation to list
2412 op_list
.append(new_op
)
2414 db_nslcmop_update
= {'_admin.operations': op_list
}
2415 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2416 op_index
= len(op_list
) - 1
2419 # Helper methods for scale() sub-operations
2421 # pre-scale/post-scale:
2422 # Check for 3 different cases:
2423 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2424 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2425 # c. Reintent: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2426 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2427 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2428 # Find this sub-operation
2429 if (RO_nsr_id
and RO_scaling_info
):
2430 operationType
= 'SCALE-RO'
2432 'member_vnf_index': vnf_index
,
2433 'RO_nsr_id': RO_nsr_id
,
2434 'RO_scaling_info': RO_scaling_info
,
2438 'member_vnf_index': vnf_index
,
2439 'primitive': vnf_config_primitive
,
2440 'primitive_params': primitive_params
,
2441 'lcmOperationType': operationType
2443 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2444 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2445 # a. New sub-operation
2446 # The sub-operation does not exist, add it.
2447 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2448 # The following parameters are set to None for all kind of scaling:
2450 vdu_count_index
= None
2452 if RO_nsr_id
and RO_scaling_info
:
2453 vnf_config_primitive
= None
2454 primitive_params
= None
2457 RO_scaling_info
= None
2458 # Initial status for sub-operation
2459 operationState
= 'PROCESSING'
2460 detailed_status
= 'In progress'
2461 # Add sub-operation for pre/post-scaling (zero or more operations)
2462 self
._add
_suboperation
(db_nslcmop
,
2467 vnf_config_primitive
,
2474 return self
.SUBOPERATION_STATUS_NEW
2476 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2477 # or op_index (operationState != 'COMPLETED')
2478 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2480 # Function to return execution_environment id
2482 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2483 # TODO vdu_index_count
2484 for vca
in vca_deployed_list
:
2485 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2488 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
, vca_index
, destroy_ee
=True):
2490 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2491 :param logging_text:
2493 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2494 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2495 :param vca_index: index in the database _admin.deployed.VCA
2496 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2497 :return: None or exception
2499 # execute terminate_primitives
2500 terminate_primitives
= config_descriptor
.get("terminate-config-primitive")
2501 vdu_id
= vca_deployed
.get("vdu_id")
2502 vdu_count_index
= vca_deployed
.get("vdu_count_index")
2503 vdu_name
= vca_deployed
.get("vdu_name")
2504 vnf_index
= vca_deployed
.get("member-vnf-index")
2505 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
2506 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2507 terminate_primitives
= sorted(terminate_primitives
, key
=lambda x
: int(x
['seq']))
2508 for seq
in terminate_primitives
:
2509 # For each sequence in list, get primitive and call _ns_execute_primitive()
2510 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
2511 vnf_index
, seq
.get("name"))
2512 self
.logger
.debug(logging_text
+ step
)
2513 # Create the primitive for each sequence, i.e. "primitive": "touch"
2514 primitive
= seq
.get('name')
2515 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
2516 # The following 3 parameters are currently set to None for 'terminate':
2517 # vdu_id, vdu_count_index, vdu_name
2520 self
._add
_suboperation
(db_nslcmop
,
2526 mapped_primitive_params
)
2527 # Sub-operations: Call _ns_execute_primitive() instead of action()
2529 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
2530 mapped_primitive_params
)
2531 except LcmException
:
2532 # this happens when VCA is not deployed. In this case it is not needed to terminate
2534 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
2535 if result
not in result_ok
:
2536 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
2537 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
2538 # set that this VCA do not need terminated
2539 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
2540 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
2543 await self
.n2vc
.delete_execution_environment(vca_deployed
["ee_id"])
2545 async def _delete_all_N2VC(self
, db_nsr
: dict):
2546 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
2547 namespace
= "." + db_nsr
["_id"]
2549 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
2550 except N2VCNotFound
: # already deleted. Skip
2552 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
2554 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
2556 Terminates a deployment from RO
2557 :param logging_text:
2558 :param nsr_deployed: db_nsr._admin.deployed
2561 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
2562 this method will update only the index 2, but it will write on database the concatenated content of the list
2567 ro_nsr_id
= ro_delete_action
= None
2568 if nsr_deployed
and nsr_deployed
.get("RO"):
2569 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
2570 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
2573 stage
[2] = "Deleting ns from VIM."
2574 db_nsr_update
["detailed-status"] = " ".join(stage
)
2575 self
._write
_op
_status
(nslcmop_id
, stage
)
2576 self
.logger
.debug(logging_text
+ stage
[2])
2577 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2578 self
._write
_op
_status
(nslcmop_id
, stage
)
2579 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
2580 ro_delete_action
= desc
["action_id"]
2581 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
2582 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2583 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2584 if ro_delete_action
:
2585 # wait until NS is deleted from VIM
2586 stage
[2] = "Waiting ns deleted from VIM."
2587 detailed_status_old
= None
2588 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
2590 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2591 self
._write
_op
_status
(nslcmop_id
, stage
)
2593 delete_timeout
= 20 * 60 # 20 minutes
2594 while delete_timeout
> 0:
2595 desc
= await self
.RO
.show(
2597 item_id_name
=ro_nsr_id
,
2598 extra_item
="action",
2599 extra_item_id
=ro_delete_action
)
2602 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
2604 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
2605 if ns_status
== "ERROR":
2606 raise ROclient
.ROClientException(ns_status_info
)
2607 elif ns_status
== "BUILD":
2608 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
2609 elif ns_status
== "ACTIVE":
2610 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2611 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2614 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
2615 if stage
[2] != detailed_status_old
:
2616 detailed_status_old
= stage
[2]
2617 db_nsr_update
["detailed-status"] = " ".join(stage
)
2618 self
._write
_op
_status
(nslcmop_id
, stage
)
2619 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2620 await asyncio
.sleep(5, loop
=self
.loop
)
2622 else: # delete_timeout <= 0:
2623 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
2625 except Exception as e
:
2626 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2627 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2628 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2629 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2630 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2631 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
2632 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2633 failed_detail
.append("delete conflict: {}".format(e
))
2634 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
2636 failed_detail
.append("delete error: {}".format(e
))
2637 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
2640 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
2641 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
2643 stage
[2] = "Deleting nsd from RO."
2644 db_nsr_update
["detailed-status"] = " ".join(stage
)
2645 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2646 self
._write
_op
_status
(nslcmop_id
, stage
)
2647 await self
.RO
.delete("nsd", ro_nsd_id
)
2648 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
2649 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2650 except Exception as e
:
2651 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2652 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2653 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
2654 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2655 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
2656 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2658 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
2659 self
.logger
.error(logging_text
+ failed_detail
[-1])
2661 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
2662 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
2663 if not vnf_deployed
or not vnf_deployed
["id"]:
2666 ro_vnfd_id
= vnf_deployed
["id"]
2667 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
2668 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
2669 db_nsr_update
["detailed-status"] = " ".join(stage
)
2670 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2671 self
._write
_op
_status
(nslcmop_id
, stage
)
2672 await self
.RO
.delete("vnfd", ro_vnfd_id
)
2673 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
2674 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2675 except Exception as e
:
2676 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2677 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2678 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
2679 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2680 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
2681 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2683 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
2684 self
.logger
.error(logging_text
+ failed_detail
[-1])
2687 stage
[2] = "Error deleting from VIM"
2689 stage
[2] = "Deleted from VIM"
2690 db_nsr_update
["detailed-status"] = " ".join(stage
)
2691 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2692 self
._write
_op
_status
(nslcmop_id
, stage
)
2695 raise LcmException("; ".join(failed_detail
))
2697 async def terminate(self
, nsr_id
, nslcmop_id
):
2698 # Try to lock HA task here
2699 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
2700 if not task_is_locked_by_me
:
2703 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
2704 self
.logger
.debug(logging_text
+ "Enter")
2705 timeout_ns_terminate
= self
.timeout_ns_terminate
2709 error_list
= [] # annotates all failed error messages
2710 db_nslcmop_update
= {}
2711 autoremove
= False # autoremove after terminated
2712 tasks_dict_info
= {}
2714 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
2715 # ^ contains [stage, step, VIM-status]
2717 # wait for any previous tasks in process
2718 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
2720 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2721 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2722 operation_params
= db_nslcmop
.get("operationParams") or {}
2723 if operation_params
.get("timeout_ns_terminate"):
2724 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
2725 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2726 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2728 db_nsr_update
["operational-status"] = "terminating"
2729 db_nsr_update
["config-status"] = "terminating"
2730 self
._write
_ns
_status
(
2732 ns_state
="TERMINATING",
2733 current_operation
="TERMINATING",
2734 current_operation_id
=nslcmop_id
,
2735 other_update
=db_nsr_update
2737 self
._write
_op
_status
(
2742 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
2743 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
2746 stage
[1] = "Getting vnf descriptors from db."
2747 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2748 db_vnfds_from_id
= {}
2749 db_vnfds_from_member_index
= {}
2751 for vnfr
in db_vnfrs_list
:
2752 vnfd_id
= vnfr
["vnfd-id"]
2753 if vnfd_id
not in db_vnfds_from_id
:
2754 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2755 db_vnfds_from_id
[vnfd_id
] = vnfd
2756 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
2758 # Destroy individual execution environments when there are terminating primitives.
2759 # Rest of EE will be deleted at once
2760 if not operation_params
.get("skip_terminate_primitives"):
2761 stage
[0] = "Stage 2/3 execute terminating primitives."
2762 stage
[1] = "Looking execution environment that needs terminate."
2763 self
.logger
.debug(logging_text
+ stage
[1])
2764 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
2765 config_descriptor
= None
2766 if not vca
or not vca
.get("ee_id") or not vca
.get("needed_terminate"):
2768 if not vca
.get("member-vnf-index"):
2770 config_descriptor
= db_nsr
.get("ns-configuration")
2771 elif vca
.get("vdu_id"):
2772 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2773 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
2775 config_descriptor
= vdud
.get("vdu-configuration")
2776 elif vca
.get("kdu_name"):
2777 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2778 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
2780 config_descriptor
= kdud
.get("kdu-configuration")
2782 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
2783 task
= asyncio
.ensure_future(self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
,
2785 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
2787 # wait for pending tasks of terminate primitives
2789 self
.logger
.debug(logging_text
+ 'Waiting for terminate primitive pending tasks...')
2790 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
2791 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
2794 return # raise LcmException("; ".join(error_list))
2795 tasks_dict_info
.clear()
2797 # remove All execution environments at once
2798 stage
[0] = "Stage 3/3 delete all."
2800 if nsr_deployed
.get("VCA"):
2801 stage
[1] = "Deleting all execution environments."
2802 self
.logger
.debug(logging_text
+ stage
[1])
2803 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
2804 timeout
=self
.timeout_charm_delete
))
2805 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
2806 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
2808 # Delete from k8scluster
2809 stage
[1] = "Deleting KDUs."
2810 self
.logger
.debug(logging_text
+ stage
[1])
2811 # print(nsr_deployed)
2812 for kdu
in get_iterable(nsr_deployed
, "K8s"):
2813 if not kdu
or not kdu
.get("kdu-instance"):
2815 kdu_instance
= kdu
.get("kdu-instance")
2816 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
2817 task_delete_kdu_instance
= asyncio
.ensure_future(
2818 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
2819 cluster_uuid
=kdu
.get("k8scluster-uuid"),
2820 kdu_instance
=kdu_instance
))
2822 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
2823 format(kdu
.get("k8scluster-type")))
2825 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
2828 stage
[1] = "Deleting ns from VIM."
2829 task_delete_ro
= asyncio
.ensure_future(
2830 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
2831 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
2833 # rest of staff will be done at finally
2835 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2836 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
2838 except asyncio
.CancelledError
:
2839 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2840 exc
= "Operation was cancelled"
2841 except Exception as e
:
2842 exc
= traceback
.format_exc()
2843 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2846 error_list
.append(str(exc
))
2848 # wait for pending tasks
2850 stage
[1] = "Waiting for terminate pending tasks."
2851 self
.logger
.debug(logging_text
+ stage
[1])
2852 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
2854 stage
[1] = stage
[2] = ""
2855 except asyncio
.CancelledError
:
2856 error_list
.append("Cancelled")
2857 # TODO cancell all tasks
2858 except Exception as exc
:
2859 error_list
.append(str(exc
))
2860 # update status at database
2862 error_detail
= "; ".join(error_list
)
2863 # self.logger.error(logging_text + error_detail)
2864 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
2865 error_description_nsr
= 'Operation: TERMINATING.{}, Stage {}.'.format(nslcmop_id
, stage
[0])
2867 db_nsr_update
["operational-status"] = "failed"
2868 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2869 db_nslcmop_update
["detailed-status"] = error_detail
2870 nslcmop_operation_state
= "FAILED"
2874 error_description_nsr
= error_description_nslcmop
= None
2875 ns_state
= "NOT_INSTANTIATED"
2876 db_nsr_update
["operational-status"] = "terminated"
2877 db_nsr_update
["detailed-status"] = "Done"
2878 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
2879 db_nslcmop_update
["detailed-status"] = "Done"
2880 nslcmop_operation_state
= "COMPLETED"
2883 self
._write
_ns
_status
(
2886 current_operation
="IDLE",
2887 current_operation_id
=None,
2888 error_description
=error_description_nsr
,
2889 error_detail
=error_detail
,
2890 other_update
=db_nsr_update
2893 self
._write
_op
_status
(
2896 error_message
=error_description_nslcmop
,
2897 operation_state
=nslcmop_operation_state
,
2898 other_update
=db_nslcmop_update
,
2900 autoremove
= operation_params
.get("autoremove", False)
2901 if nslcmop_operation_state
:
2903 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2904 "operationState": nslcmop_operation_state
,
2905 "autoremove": autoremove
},
2907 except Exception as e
:
2908 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2910 self
.logger
.debug(logging_text
+ "Exit")
2911 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
2913 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
2915 error_detail_list
= []
2917 pending_tasks
= list(created_tasks_info
.keys())
2918 num_tasks
= len(pending_tasks
)
2920 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
2921 self
._write
_op
_status
(nslcmop_id
, stage
)
2922 while pending_tasks
:
2924 _timeout
= timeout
+ time_start
- time()
2925 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
2926 return_when
=asyncio
.FIRST_COMPLETED
)
2927 num_done
+= len(done
)
2928 if not done
: # Timeout
2929 for task
in pending_tasks
:
2930 new_error
= created_tasks_info
[task
] + ": Timeout"
2931 error_detail_list
.append(new_error
)
2932 error_list
.append(new_error
)
2935 if task
.cancelled():
2938 exc
= task
.exception()
2940 if isinstance(exc
, asyncio
.TimeoutError
):
2942 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
2943 error_list
.append(created_tasks_info
[task
])
2944 error_detail_list
.append(new_error
)
2945 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
)):
2946 self
.logger
.error(logging_text
+ new_error
)
2948 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
2949 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
2951 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
2952 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
2954 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
2955 if nsr_id
: # update also nsr
2956 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
2957 "errorDetail": ". ".join(error_detail_list
)})
2958 self
._write
_op
_status
(nslcmop_id
, stage
)
2959 return error_detail_list
2962 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
2964 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
2965 The default-value is used. If it is between < > it look for a value at instantiation_params
2966 :param primitive_desc: portion of VNFD/NSD that describes primitive
2967 :param params: Params provided by user
2968 :param instantiation_params: Instantiation params provided by user
2969 :return: a dictionary with the calculated params
2971 calculated_params
= {}
2972 for parameter
in primitive_desc
.get("parameter", ()):
2973 param_name
= parameter
["name"]
2974 if param_name
in params
:
2975 calculated_params
[param_name
] = params
[param_name
]
2976 elif "default-value" in parameter
or "value" in parameter
:
2977 if "value" in parameter
:
2978 calculated_params
[param_name
] = parameter
["value"]
2980 calculated_params
[param_name
] = parameter
["default-value"]
2981 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
2982 and calculated_params
[param_name
].endswith(">"):
2983 if calculated_params
[param_name
][1:-1] in instantiation_params
:
2984 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
2986 raise LcmException("Parameter {} needed to execute primitive {} not provided".
2987 format(calculated_params
[param_name
], primitive_desc
["name"]))
2989 raise LcmException("Parameter {} needed to execute primitive {} not provided".
2990 format(param_name
, primitive_desc
["name"]))
2992 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
2993 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
2995 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
2996 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
2998 # add always ns_config_info if primitive name is config
2999 if primitive_desc
["name"] == "config":
3000 if "ns_config_info" in instantiation_params
:
3001 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3002 return calculated_params
3004 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None):
3005 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3006 for vca
in deployed_vca
:
3009 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3011 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3013 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3017 # vca_deployed not found
3018 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} is not "
3019 "deployed".format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3022 ee_id
= vca
.get("ee_id")
3024 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3025 "execution environment"
3026 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3029 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3030 retries_interval
=30, timeout
=None) -> (str, str):
3032 if primitive
== "config":
3033 primitive_params
= {"params": primitive_params
}
3037 output
= await asyncio
.wait_for(
3038 self
.n2vc
.exec_primitive(
3040 primitive_name
=primitive
,
3041 params_dict
=primitive_params
,
3042 progress_timeout
=self
.timeout_progress_primitive
,
3043 total_timeout
=self
.timeout_primitive
),
3044 timeout
=timeout
or self
.timeout_primitive
)
3047 except asyncio
.CancelledError
:
3049 except Exception as e
: # asyncio.TimeoutError
3050 if isinstance(e
, asyncio
.TimeoutError
):
3054 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3056 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3058 return 'FAILED', str(e
)
3060 return 'COMPLETED', output
3062 except (LcmException
, asyncio
.CancelledError
):
3064 except Exception as e
:
3065 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3067 async def action(self
, nsr_id
, nslcmop_id
):
3069 # Try to lock HA task here
3070 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3071 if not task_is_locked_by_me
:
3074 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3075 self
.logger
.debug(logging_text
+ "Enter")
3076 # get all needed from database
3080 db_nslcmop_update
= {}
3081 nslcmop_operation_state
= None
3082 error_description_nslcmop
= None
3085 # wait for any previous tasks in process
3086 step
= "Waiting for previous operations to terminate"
3087 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3089 self
._write
_ns
_status
(
3092 current_operation
="RUNNING ACTION",
3093 current_operation_id
=nslcmop_id
3096 step
= "Getting information from database"
3097 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3098 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3100 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3101 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3102 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3103 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3104 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3105 primitive
= db_nslcmop
["operationParams"]["primitive"]
3106 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3107 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3110 step
= "Getting vnfr from database"
3111 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3112 step
= "Getting vnfd from database"
3113 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3115 step
= "Getting nsd from database"
3116 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3118 # for backward compatibility
3119 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3120 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3121 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3122 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3124 # look for primitive
3125 config_primitive_desc
= None
3127 for vdu
in get_iterable(db_vnfd
, "vdu"):
3128 if vdu_id
== vdu
["id"]:
3129 for config_primitive
in deep_get(vdu
, ("vdu-configuration", "config-primitive"), ()):
3130 if config_primitive
["name"] == primitive
:
3131 config_primitive_desc
= config_primitive
3135 for kdu
in get_iterable(db_vnfd
, "kdu"):
3136 if kdu_name
== kdu
["name"]:
3137 for config_primitive
in deep_get(kdu
, ("kdu-configuration", "config-primitive"), ()):
3138 if config_primitive
["name"] == primitive
:
3139 config_primitive_desc
= config_primitive
3143 for config_primitive
in deep_get(db_vnfd
, ("vnf-configuration", "config-primitive"), ()):
3144 if config_primitive
["name"] == primitive
:
3145 config_primitive_desc
= config_primitive
3148 for config_primitive
in deep_get(db_nsd
, ("ns-configuration", "config-primitive"), ()):
3149 if config_primitive
["name"] == primitive
:
3150 config_primitive_desc
= config_primitive
3153 if not config_primitive_desc
and not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3154 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3159 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3160 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3162 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3163 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3165 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3167 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3169 # TODO check if ns is in a proper status
3170 if kdu_name
and primitive
in ("upgrade", "rollback", "status"):
3171 # kdur and desc_params already set from before
3172 if primitive_params
:
3173 desc_params
.update(primitive_params
)
3174 # TODO Check if we will need something at vnf level
3175 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3176 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3179 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3181 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3182 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3183 raise LcmException(msg
)
3185 db_dict
= {"collection": "nsrs",
3186 "filter": {"_id": nsr_id
},
3187 "path": "_admin.deployed.K8s.{}".format(index
)}
3188 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive
, vnf_index
, kdu_name
))
3189 step
= "Executing kdu {}".format(primitive
)
3190 if primitive
== "upgrade":
3191 if desc_params
.get("kdu_model"):
3192 kdu_model
= desc_params
.get("kdu_model")
3193 del desc_params
["kdu_model"]
3195 kdu_model
= kdu
.get("kdu-model")
3196 parts
= kdu_model
.split(sep
=":")
3198 kdu_model
= parts
[0]
3200 detailed_status
= await asyncio
.wait_for(
3201 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3202 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3203 kdu_instance
=kdu
.get("kdu-instance"),
3204 atomic
=True, kdu_model
=kdu_model
,
3205 params
=desc_params
, db_dict
=db_dict
,
3206 timeout
=timeout_ns_action
),
3207 timeout
=timeout_ns_action
+ 10)
3208 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3209 elif primitive
== "rollback":
3210 detailed_status
= await asyncio
.wait_for(
3211 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3212 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3213 kdu_instance
=kdu
.get("kdu-instance"),
3215 timeout
=timeout_ns_action
)
3216 elif primitive
== "status":
3217 detailed_status
= await asyncio
.wait_for(
3218 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3219 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3220 kdu_instance
=kdu
.get("kdu-instance")),
3221 timeout
=timeout_ns_action
)
3224 nslcmop_operation_state
= 'COMPLETED'
3226 detailed_status
= ''
3227 nslcmop_operation_state
= 'FAILED'
3230 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3231 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3232 member_vnf_index
=vnf_index
,
3234 vdu_count_index
=vdu_count_index
),
3235 primitive
=primitive
,
3236 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3237 timeout
=timeout_ns_action
)
3239 db_nslcmop_update
["detailed-status"] = detailed_status
3240 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3241 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3243 return # database update is called inside finally
3245 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3246 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3248 except asyncio
.CancelledError
:
3249 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3250 exc
= "Operation was cancelled"
3251 except asyncio
.TimeoutError
:
3252 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3254 except Exception as e
:
3255 exc
= traceback
.format_exc()
3256 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3259 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3260 "FAILED {}: {}".format(step
, exc
)
3261 nslcmop_operation_state
= "FAILED"
3263 self
._write
_ns
_status
(
3265 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3266 current_operation
="IDLE",
3267 current_operation_id
=None,
3268 # error_description=error_description_nsr,
3269 # error_detail=error_detail,
3270 other_update
=db_nsr_update
3274 self
._write
_op
_status
(
3277 error_message
=error_description_nslcmop
,
3278 operation_state
=nslcmop_operation_state
,
3279 other_update
=db_nslcmop_update
,
3282 if nslcmop_operation_state
:
3284 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3285 "operationState": nslcmop_operation_state
},
3287 except Exception as e
:
3288 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3289 self
.logger
.debug(logging_text
+ "Exit")
3290 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3291 return nslcmop_operation_state
, detailed_status
3293 async def scale(self
, nsr_id
, nslcmop_id
):
3295 # Try to lock HA task here
3296 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3297 if not task_is_locked_by_me
:
3300 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3301 self
.logger
.debug(logging_text
+ "Enter")
3302 # get all needed from database
3305 db_nslcmop_update
= {}
3306 nslcmop_operation_state
= None
3309 # in case of error, indicates what part of scale was failed to put nsr at error status
3310 scale_process
= None
3311 old_operational_status
= ""
3312 old_config_status
= ""
3315 # wait for any previous tasks in process
3316 step
= "Waiting for previous operations to terminate"
3317 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3319 self
._write
_ns
_status
(
3322 current_operation
="SCALING",
3323 current_operation_id
=nslcmop_id
3326 step
= "Getting nslcmop from database"
3327 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3328 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3329 step
= "Getting nsr from database"
3330 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3332 old_operational_status
= db_nsr
["operational-status"]
3333 old_config_status
= db_nsr
["config-status"]
3334 step
= "Parsing scaling parameters"
3335 # self.logger.debug(step)
3336 db_nsr_update
["operational-status"] = "scaling"
3337 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3338 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3341 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3342 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3343 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3344 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3345 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3348 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3349 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3350 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3351 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3352 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3354 # for backward compatibility
3355 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3356 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3357 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3358 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3360 step
= "Getting vnfr from database"
3361 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3362 step
= "Getting vnfd from database"
3363 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3365 step
= "Getting scaling-group-descriptor"
3366 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3367 if scaling_descriptor
["name"] == scaling_group
:
3370 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3371 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3374 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3375 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3376 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3379 # TODO check if ns is in a proper status
3380 step
= "Sending scale order to VIM"
3382 if not db_nsr
["_admin"].get("scaling-group"):
3383 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3384 admin_scale_index
= 0
3386 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3387 if admin_scale_info
["name"] == scaling_group
:
3388 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3390 else: # not found, set index one plus last element and add new entry with the name
3391 admin_scale_index
+= 1
3392 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3393 RO_scaling_info
= []
3394 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3395 if scaling_type
== "SCALE_OUT":
3396 # count if max-instance-count is reached
3397 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3398 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3399 if nb_scale_op
>= max_instance_count
:
3400 raise LcmException("reached the limit of {} (max-instance-count) "
3401 "scaling-out operations for the "
3402 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3405 vdu_scaling_info
["scaling_direction"] = "OUT"
3406 vdu_scaling_info
["vdu-create"] = {}
3407 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3408 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3409 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3410 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3412 elif scaling_type
== "SCALE_IN":
3413 # count if min-instance-count is reached
3414 min_instance_count
= 0
3415 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3416 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3417 if nb_scale_op
<= min_instance_count
:
3418 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3419 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3421 vdu_scaling_info
["scaling_direction"] = "IN"
3422 vdu_scaling_info
["vdu-delete"] = {}
3423 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3424 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3425 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3426 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3428 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3429 vdu_create
= vdu_scaling_info
.get("vdu-create")
3430 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3431 if vdu_scaling_info
["scaling_direction"] == "IN":
3432 for vdur
in reversed(db_vnfr
["vdur"]):
3433 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3434 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3435 vdu_scaling_info
["vdu"].append({
3436 "name": vdur
["name"],
3437 "vdu_id": vdur
["vdu-id-ref"],
3440 for interface
in vdur
["interfaces"]:
3441 vdu_scaling_info
["vdu"][-1]["interface"].append({
3442 "name": interface
["name"],
3443 "ip_address": interface
["ip-address"],
3444 "mac_address": interface
.get("mac-address"),
3446 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
3449 step
= "Executing pre-scale vnf-config-primitive"
3450 if scaling_descriptor
.get("scaling-config-action"):
3451 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3452 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
3453 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
3454 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3455 step
= db_nslcmop_update
["detailed-status"] = \
3456 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3458 # look for primitive
3459 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3460 if config_primitive
["name"] == vnf_config_primitive
:
3464 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
3465 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
3466 "primitive".format(scaling_group
, config_primitive
))
3468 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3469 if db_vnfr
.get("additionalParamsForVnf"):
3470 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3472 scale_process
= "VCA"
3473 db_nsr_update
["config-status"] = "configuring pre-scaling"
3474 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3476 # Pre-scale reintent check: Check if this sub-operation has been executed before
3477 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3478 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
3479 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3480 # Skip sub-operation
3481 result
= 'COMPLETED'
3482 result_detail
= 'Done'
3483 self
.logger
.debug(logging_text
+
3484 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
3485 vnf_config_primitive
, result
, result_detail
))
3487 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3488 # New sub-operation: Get index of this sub-operation
3489 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3490 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3491 format(vnf_config_primitive
))
3493 # Reintent: Get registered params for this existing sub-operation
3494 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3495 vnf_index
= op
.get('member_vnf_index')
3496 vnf_config_primitive
= op
.get('primitive')
3497 primitive_params
= op
.get('primitive_params')
3498 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3499 format(vnf_config_primitive
))
3500 # Execute the primitive, either with new (first-time) or registered (reintent) args
3501 result
, result_detail
= await self
._ns
_execute
_primitive
(
3502 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3503 member_vnf_index
=vnf_index
,
3505 vdu_count_index
=None),
3506 vnf_config_primitive
, primitive_params
)
3507 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3508 vnf_config_primitive
, result
, result_detail
))
3509 # Update operationState = COMPLETED | FAILED
3510 self
._update
_suboperation
_status
(
3511 db_nslcmop
, op_index
, result
, result_detail
)
3513 if result
== "FAILED":
3514 raise LcmException(result_detail
)
3515 db_nsr_update
["config-status"] = old_config_status
3516 scale_process
= None
3520 # Should this block be skipped if 'RO_nsr_id' == None ?
3521 # if (RO_nsr_id and RO_scaling_info):
3523 scale_process
= "RO"
3524 # Scale RO reintent check: Check if this sub-operation has been executed before
3525 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3526 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
3527 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3528 # Skip sub-operation
3529 result
= 'COMPLETED'
3530 result_detail
= 'Done'
3531 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
3532 result
, result_detail
))
3534 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3535 # New sub-operation: Get index of this sub-operation
3536 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3537 self
.logger
.debug(logging_text
+ "New sub-operation RO")
3539 # Reintent: Get registered params for this existing sub-operation
3540 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3541 RO_nsr_id
= op
.get('RO_nsr_id')
3542 RO_scaling_info
= op
.get('RO_scaling_info')
3543 self
.logger
.debug(logging_text
+ "Sub-operation RO reintent".format(
3544 vnf_config_primitive
))
3546 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
3547 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
3548 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
3550 RO_nslcmop_id
= RO_desc
["instance_action_id"]
3551 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
3553 RO_task_done
= False
3554 step
= detailed_status
= "Waiting RO_task_id={} to complete the scale action.".format(RO_nslcmop_id
)
3555 detailed_status_old
= None
3556 self
.logger
.debug(logging_text
+ step
)
3558 deployment_timeout
= 1 * 3600 # One hour
3559 while deployment_timeout
> 0:
3560 if not RO_task_done
:
3561 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
3562 extra_item_id
=RO_nslcmop_id
)
3565 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3567 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3568 if ns_status
== "ERROR":
3569 raise ROclient
.ROClientException(ns_status_info
)
3570 elif ns_status
== "BUILD":
3571 detailed_status
= step
+ "; {}".format(ns_status_info
)
3572 elif ns_status
== "ACTIVE":
3574 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
3575 self
.logger
.debug(logging_text
+ step
)
3577 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3580 if ns_status
== "ERROR":
3581 raise ROclient
.ROClientException(ns_status_info
)
3582 elif ns_status
== "BUILD":
3583 detailed_status
= step
+ "; {}".format(ns_status_info
)
3584 elif ns_status
== "ACTIVE":
3585 step
= detailed_status
= \
3586 "Waiting for management IP address reported by the VIM. Updating VNFRs"
3588 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
3591 desc
= await self
.RO
.show("ns", RO_nsr_id
)
3594 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3596 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
3597 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
3599 except LcmExceptionNoMgmtIP
:
3602 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
3603 if detailed_status
!= detailed_status_old
:
3604 self
._update
_suboperation
_status
(
3605 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
3606 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
3607 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3609 await asyncio
.sleep(5, loop
=self
.loop
)
3610 deployment_timeout
-= 5
3611 if deployment_timeout
<= 0:
3612 self
._update
_suboperation
_status
(
3613 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
3614 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
3616 # update VDU_SCALING_INFO with the obtained ip_addresses
3617 if vdu_scaling_info
["scaling_direction"] == "OUT":
3618 for vdur
in reversed(db_vnfr
["vdur"]):
3619 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
3620 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
3621 vdu_scaling_info
["vdu"].append({
3622 "name": vdur
["name"],
3623 "vdu_id": vdur
["vdu-id-ref"],
3626 for interface
in vdur
["interfaces"]:
3627 vdu_scaling_info
["vdu"][-1]["interface"].append({
3628 "name": interface
["name"],
3629 "ip_address": interface
["ip-address"],
3630 "mac_address": interface
.get("mac-address"),
3632 del vdu_scaling_info
["vdu-create"]
3634 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
3637 scale_process
= None
3639 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3642 # execute primitive service POST-SCALING
3643 step
= "Executing post-scale vnf-config-primitive"
3644 if scaling_descriptor
.get("scaling-config-action"):
3645 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3646 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
3647 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
3648 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3649 step
= db_nslcmop_update
["detailed-status"] = \
3650 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3652 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3653 if db_vnfr
.get("additionalParamsForVnf"):
3654 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3656 # look for primitive
3657 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3658 if config_primitive
["name"] == vnf_config_primitive
:
3661 raise LcmException("Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:"
3662 "scaling-config-action[vnf-config-primitive-name-ref='{}'] does not "
3663 "match any vnf-configuration:config-primitive".format(scaling_group
,
3665 scale_process
= "VCA"
3666 db_nsr_update
["config-status"] = "configuring post-scaling"
3667 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3669 # Post-scale reintent check: Check if this sub-operation has been executed before
3670 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3671 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
3672 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3673 # Skip sub-operation
3674 result
= 'COMPLETED'
3675 result_detail
= 'Done'
3676 self
.logger
.debug(logging_text
+
3677 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
3678 format(vnf_config_primitive
, result
, result_detail
))
3680 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3681 # New sub-operation: Get index of this sub-operation
3682 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3683 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3684 format(vnf_config_primitive
))
3686 # Reintent: Get registered params for this existing sub-operation
3687 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3688 vnf_index
= op
.get('member_vnf_index')
3689 vnf_config_primitive
= op
.get('primitive')
3690 primitive_params
= op
.get('primitive_params')
3691 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3692 format(vnf_config_primitive
))
3693 # Execute the primitive, either with new (first-time) or registered (reintent) args
3694 result
, result_detail
= await self
._ns
_execute
_primitive
(
3695 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3696 member_vnf_index
=vnf_index
,
3698 vdu_count_index
=None),
3699 vnf_config_primitive
, primitive_params
)
3700 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3701 vnf_config_primitive
, result
, result_detail
))
3702 # Update operationState = COMPLETED | FAILED
3703 self
._update
_suboperation
_status
(
3704 db_nslcmop
, op_index
, result
, result_detail
)
3706 if result
== "FAILED":
3707 raise LcmException(result_detail
)
3708 db_nsr_update
["config-status"] = old_config_status
3709 scale_process
= None
3712 db_nslcmop_update
["operationState"] = nslcmop_operation_state
= "COMPLETED"
3713 db_nslcmop_update
["statusEnteredTime"] = time()
3714 db_nslcmop_update
["detailed-status"] = "done"
3715 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
3716 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
3717 else old_operational_status
3718 db_nsr_update
["config-status"] = old_config_status
3720 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
3721 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3723 except asyncio
.CancelledError
:
3724 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3725 exc
= "Operation was cancelled"
3726 except Exception as e
:
3727 exc
= traceback
.format_exc()
3728 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3730 self
._write
_ns
_status
(
3733 current_operation
="IDLE",
3734 current_operation_id
=None
3738 db_nslcmop_update
["detailed-status"] = "FAILED {}: {}".format(step
, exc
)
3739 db_nslcmop_update
["operationState"] = nslcmop_operation_state
= "FAILED"
3740 db_nslcmop_update
["statusEnteredTime"] = time()
3742 db_nsr_update
["operational-status"] = old_operational_status
3743 db_nsr_update
["config-status"] = old_config_status
3744 db_nsr_update
["detailed-status"] = ""
3746 if "VCA" in scale_process
:
3747 db_nsr_update
["config-status"] = "failed"
3748 if "RO" in scale_process
:
3749 db_nsr_update
["operational-status"] = "failed"
3750 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
3753 if db_nslcmop
and db_nslcmop_update
:
3754 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3756 self
._write
_ns
_status
(
3759 current_operation
="IDLE",
3760 current_operation_id
=None,
3761 other_update
=db_nsr_update
3764 except DbException
as e
:
3765 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
3766 if nslcmop_operation_state
:
3768 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3769 "operationState": nslcmop_operation_state
},
3772 # await asyncio.sleep(cooldown_time, loop=self.loop)
3773 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
3774 except Exception as e
:
3775 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3776 self
.logger
.debug(logging_text
+ "Exit")
3777 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")