1 # -*- coding: utf-8 -*-
4 # Copyright 2018 Telefonica S.A.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
22 import logging
.handlers
25 from jinja2
import Environment
, Template
, meta
, TemplateError
, TemplateNotFound
, TemplateSyntaxError
27 from osm_lcm
import ROclient
28 from osm_lcm
.lcm_utils
import LcmException
, LcmExceptionNoMgmtIP
, LcmBase
, deep_get
, get_iterable
, populate_dict
29 from n2vc
.k8s_helm_conn
import K8sHelmConnector
30 from n2vc
.k8s_juju_conn
import K8sJujuConnector
32 from osm_common
.dbbase
import DbException
33 from osm_common
.fsbase
import FsException
35 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
36 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
38 from copy
import copy
, deepcopy
39 from http
import HTTPStatus
41 from uuid
import uuid4
42 from functools
import partial
44 __author__
= "Alfonso Tierno"
48 timeout_vca_on_error
= 5 * 60 # Time for charm from first time at blocked,error status to mark as failed
49 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
50 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
51 timeout_charm_delete
= 10 * 60
52 timeout_primitive
= 10 * 60 # timeout for primitive execution
53 timeout_progress_primitive
= 2 * 60 # timeout for some progress in a primitive execution
55 SUBOPERATION_STATUS_NOT_FOUND
= -1
56 SUBOPERATION_STATUS_NEW
= -2
57 SUBOPERATION_STATUS_SKIP
= -3
58 task_name_deploy_vca
= "Deploying VCA"
60 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
62 Init, Connect to database, filesystem storage, and messaging
63 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
70 logger
=logging
.getLogger('lcm.ns')
74 self
.lcm_tasks
= lcm_tasks
75 self
.timeout
= config
["timeout"]
76 self
.ro_config
= config
["ro_config"]
77 self
.vca_config
= config
["VCA"].copy()
79 # create N2VC connector
80 self
.n2vc
= N2VCJujuConnector(
85 url
='{}:{}'.format(self
.vca_config
['host'], self
.vca_config
['port']),
86 username
=self
.vca_config
.get('user', None),
87 vca_config
=self
.vca_config
,
88 on_update_db
=self
._on
_update
_n
2vc
_db
91 self
.k8sclusterhelm
= K8sHelmConnector(
92 kubectl_command
=self
.vca_config
.get("kubectlpath"),
93 helm_command
=self
.vca_config
.get("helmpath"),
100 self
.k8sclusterjuju
= K8sJujuConnector(
101 kubectl_command
=self
.vca_config
.get("kubectlpath"),
102 juju_command
=self
.vca_config
.get("jujupath"),
109 self
.k8scluster_map
= {
110 "helm-chart": self
.k8sclusterhelm
,
111 "chart": self
.k8sclusterhelm
,
112 "juju-bundle": self
.k8sclusterjuju
,
113 "juju": self
.k8sclusterjuju
,
116 self
.RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
118 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
120 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
123 # TODO filter RO descriptor fields...
127 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
128 db_dict
['deploymentStatus'] = ro_descriptor
129 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
131 except Exception as e
:
132 self
.logger
.warn('Cannot write database RO deployment for ns={} -> {}'.format(nsrs_id
, e
))
134 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
):
136 # remove last dot from path (if exists)
137 if path
.endswith('.'):
140 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
141 # .format(table, filter, path, updated_data))
145 nsr_id
= filter.get('_id')
147 # read ns record from database
148 nsr
= self
.db
.get_one(table
='nsrs', q_filter
=filter)
149 current_ns_status
= nsr
.get('nsState')
151 # get vca status for NS
152 status_dict
= await self
.n2vc
.get_status(namespace
='.' + nsr_id
, yaml_format
=False)
156 db_dict
['vcaStatus'] = status_dict
158 # update configurationStatus for this VCA
160 vca_index
= int(path
[path
.rfind(".")+1:])
162 vca_list
= deep_get(target_dict
=nsr
, key_list
=('_admin', 'deployed', 'VCA'))
163 vca_status
= vca_list
[vca_index
].get('status')
165 configuration_status_list
= nsr
.get('configurationStatus')
166 config_status
= configuration_status_list
[vca_index
].get('status')
168 if config_status
== 'BROKEN' and vca_status
!= 'failed':
169 db_dict
['configurationStatus'][vca_index
] = 'READY'
170 elif config_status
!= 'BROKEN' and vca_status
== 'failed':
171 db_dict
['configurationStatus'][vca_index
] = 'BROKEN'
172 except Exception as e
:
173 # not update configurationStatus
174 self
.logger
.debug('Error updating vca_index (ignore): {}'.format(e
))
176 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
177 # if nsState = 'DEGRADED' check if all is OK
179 if current_ns_status
in ('READY', 'DEGRADED'):
180 error_description
= ''
182 if status_dict
.get('machines'):
183 for machine_id
in status_dict
.get('machines'):
184 machine
= status_dict
.get('machines').get(machine_id
)
185 # check machine agent-status
186 if machine
.get('agent-status'):
187 s
= machine
.get('agent-status').get('status')
190 error_description
+= 'machine {} agent-status={} ; '.format(machine_id
, s
)
191 # check machine instance status
192 if machine
.get('instance-status'):
193 s
= machine
.get('instance-status').get('status')
196 error_description
+= 'machine {} instance-status={} ; '.format(machine_id
, s
)
198 if status_dict
.get('applications'):
199 for app_id
in status_dict
.get('applications'):
200 app
= status_dict
.get('applications').get(app_id
)
201 # check application status
202 if app
.get('status'):
203 s
= app
.get('status').get('status')
206 error_description
+= 'application {} status={} ; '.format(app_id
, s
)
208 if error_description
:
209 db_dict
['errorDescription'] = error_description
210 if current_ns_status
== 'READY' and is_degraded
:
211 db_dict
['nsState'] = 'DEGRADED'
212 if current_ns_status
== 'DEGRADED' and not is_degraded
:
213 db_dict
['nsState'] = 'READY'
216 self
.update_db_2("nsrs", nsr_id
, db_dict
)
218 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
220 except Exception as e
:
221 self
.logger
.warn('Error updating NS state for ns={}: {}'.format(nsr_id
, e
))
223 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
225 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
226 :param vnfd: input vnfd
227 :param new_id: overrides vnf id if provided
228 :param additionalParams: Instantiation params for VNFs provided
229 :param nsrId: Id of the NSR
230 :return: copy of vnfd
233 vnfd_RO
= deepcopy(vnfd
)
234 # remove unused by RO configuration, monitoring, scaling and internal keys
235 vnfd_RO
.pop("_id", None)
236 vnfd_RO
.pop("_admin", None)
237 vnfd_RO
.pop("vnf-configuration", None)
238 vnfd_RO
.pop("monitoring-param", None)
239 vnfd_RO
.pop("scaling-group-descriptor", None)
240 vnfd_RO
.pop("kdu", None)
241 vnfd_RO
.pop("k8s-cluster", None)
243 vnfd_RO
["id"] = new_id
245 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
246 for vdu
in get_iterable(vnfd_RO
, "vdu"):
247 cloud_init_file
= None
248 if vdu
.get("cloud-init-file"):
249 base_folder
= vnfd
["_admin"]["storage"]
250 cloud_init_file
= "{}/{}/cloud_init/{}".format(base_folder
["folder"], base_folder
["pkg-dir"],
251 vdu
["cloud-init-file"])
252 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
253 cloud_init_content
= ci_file
.read()
254 vdu
.pop("cloud-init-file", None)
255 elif vdu
.get("cloud-init"):
256 cloud_init_content
= vdu
["cloud-init"]
261 ast
= env
.parse(cloud_init_content
)
262 mandatory_vars
= meta
.find_undeclared_variables(ast
)
264 for var
in mandatory_vars
:
265 if not additionalParams
or var
not in additionalParams
.keys():
266 raise LcmException("Variable '{}' defined at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
267 "file, must be provided in the instantiation parameters inside the "
268 "'additionalParamsForVnf' block".format(var
, vnfd
["id"], vdu
["id"]))
269 template
= Template(cloud_init_content
)
270 cloud_init_content
= template
.render(additionalParams
or {})
271 vdu
["cloud-init"] = cloud_init_content
274 except FsException
as e
:
275 raise LcmException("Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".
276 format(vnfd
["id"], vdu
["id"], cloud_init_file
, e
))
277 except (TemplateError
, TemplateNotFound
, TemplateSyntaxError
) as e
:
278 raise LcmException("Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".
279 format(vnfd
["id"], vdu
["id"], e
))
281 def _ns_params_2_RO(self
, ns_params
, nsd
, vnfd_dict
, db_vnfrs
, n2vc_key_list
):
283 Creates a RO ns descriptor from OSM ns_instantiate params
284 :param ns_params: OSM instantiate params
285 :param vnfd_dict: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
286 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index. {member-vnf-index: {vnfr_object}, ...}
287 :return: The RO ns descriptor
291 # TODO feature 1417: Check that no instantiation is set over PDU
292 # check if PDU forces a concrete vim-network-id and add it
293 # check if PDU contains a SDN-assist info (dpid, switch, port) and pass it to RO
295 def vim_account_2_RO(vim_account
):
296 if vim_account
in vim_2_RO
:
297 return vim_2_RO
[vim_account
]
299 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
300 if db_vim
["_admin"]["operationalState"] != "ENABLED":
301 raise LcmException("VIM={} is not available. operationalState={}".format(
302 vim_account
, db_vim
["_admin"]["operationalState"]))
303 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
304 vim_2_RO
[vim_account
] = RO_vim_id
307 def wim_account_2_RO(wim_account
):
308 if isinstance(wim_account
, str):
309 if wim_account
in wim_2_RO
:
310 return wim_2_RO
[wim_account
]
312 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
313 if db_wim
["_admin"]["operationalState"] != "ENABLED":
314 raise LcmException("WIM={} is not available. operationalState={}".format(
315 wim_account
, db_wim
["_admin"]["operationalState"]))
316 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
317 wim_2_RO
[wim_account
] = RO_wim_id
322 def ip_profile_2_RO(ip_profile
):
323 RO_ip_profile
= deepcopy((ip_profile
))
324 if "dns-server" in RO_ip_profile
:
325 if isinstance(RO_ip_profile
["dns-server"], list):
326 RO_ip_profile
["dns-address"] = []
327 for ds
in RO_ip_profile
.pop("dns-server"):
328 RO_ip_profile
["dns-address"].append(ds
['address'])
330 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
331 if RO_ip_profile
.get("ip-version") == "ipv4":
332 RO_ip_profile
["ip-version"] = "IPv4"
333 if RO_ip_profile
.get("ip-version") == "ipv6":
334 RO_ip_profile
["ip-version"] = "IPv6"
335 if "dhcp-params" in RO_ip_profile
:
336 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
342 # "name": ns_params["nsName"],
343 # "description": ns_params.get("nsDescription"),
344 "datacenter": vim_account_2_RO(ns_params
["vimAccountId"]),
345 "wim_account": wim_account_2_RO(ns_params
.get("wimAccountId")),
346 # "scenario": ns_params["nsdId"],
348 # set vim_account of each vnf if different from general vim_account.
349 # Get this information from <vnfr> database content, key vim-account-id
350 # Vim account can be set by placement_engine and it may be different from
351 # the instantiate parameters (vnfs.member-vnf-index.datacenter).
352 for vnf_index
, vnfr
in db_vnfrs
.items():
353 if vnfr
.get("vim-account-id") and vnfr
["vim-account-id"] != ns_params
["vimAccountId"]:
354 populate_dict(RO_ns_params
, ("vnfs", vnf_index
, "datacenter"), vim_account_2_RO(vnfr
["vim-account-id"]))
356 n2vc_key_list
= n2vc_key_list
or []
357 for vnfd_ref
, vnfd
in vnfd_dict
.items():
358 vdu_needed_access
= []
360 if vnfd
.get("vnf-configuration"):
361 ssh_required
= deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required"))
362 if ssh_required
and vnfd
.get("mgmt-interface"):
363 if vnfd
["mgmt-interface"].get("vdu-id"):
364 vdu_needed_access
.append(vnfd
["mgmt-interface"]["vdu-id"])
365 elif vnfd
["mgmt-interface"].get("cp"):
366 mgmt_cp
= vnfd
["mgmt-interface"]["cp"]
368 for vdu
in vnfd
.get("vdu", ()):
369 if vdu
.get("vdu-configuration"):
370 ssh_required
= deep_get(vdu
, ("vdu-configuration", "config-access", "ssh-access", "required"))
372 vdu_needed_access
.append(vdu
["id"])
374 for vdu_interface
in vdu
.get("interface"):
375 if vdu_interface
.get("external-connection-point-ref") and \
376 vdu_interface
["external-connection-point-ref"] == mgmt_cp
:
377 vdu_needed_access
.append(vdu
["id"])
381 if vdu_needed_access
:
382 for vnf_member
in nsd
.get("constituent-vnfd"):
383 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
385 for vdu
in vdu_needed_access
:
386 populate_dict(RO_ns_params
,
387 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
, "mgmt_keys"),
390 if ns_params
.get("vduImage"):
391 RO_ns_params
["vduImage"] = ns_params
["vduImage"]
393 if ns_params
.get("ssh_keys"):
394 RO_ns_params
["cloud-config"] = {"key-pairs": ns_params
["ssh_keys"]}
395 for vnf_params
in get_iterable(ns_params
, "vnf"):
396 for constituent_vnfd
in nsd
["constituent-vnfd"]:
397 if constituent_vnfd
["member-vnf-index"] == vnf_params
["member-vnf-index"]:
398 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
401 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index={} is not present at nsd:"
402 "constituent-vnfd".format(vnf_params
["member-vnf-index"]))
404 for vdu_params
in get_iterable(vnf_params
, "vdu"):
405 # TODO feature 1417: check that this VDU exist and it is not a PDU
406 if vdu_params
.get("volume"):
407 for volume_params
in vdu_params
["volume"]:
408 if volume_params
.get("vim-volume-id"):
409 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
410 vdu_params
["id"], "devices", volume_params
["name"], "vim_id"),
411 volume_params
["vim-volume-id"])
412 if vdu_params
.get("interface"):
413 for interface_params
in vdu_params
["interface"]:
414 if interface_params
.get("ip-address"):
415 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
416 vdu_params
["id"], "interfaces", interface_params
["name"],
418 interface_params
["ip-address"])
419 if interface_params
.get("mac-address"):
420 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
421 vdu_params
["id"], "interfaces", interface_params
["name"],
423 interface_params
["mac-address"])
424 if interface_params
.get("floating-ip-required"):
425 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
426 vdu_params
["id"], "interfaces", interface_params
["name"],
428 interface_params
["floating-ip-required"])
430 for internal_vld_params
in get_iterable(vnf_params
, "internal-vld"):
431 if internal_vld_params
.get("vim-network-name"):
432 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
433 internal_vld_params
["name"], "vim-network-name"),
434 internal_vld_params
["vim-network-name"])
435 if internal_vld_params
.get("vim-network-id"):
436 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
437 internal_vld_params
["name"], "vim-network-id"),
438 internal_vld_params
["vim-network-id"])
439 if internal_vld_params
.get("ip-profile"):
440 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
441 internal_vld_params
["name"], "ip-profile"),
442 ip_profile_2_RO(internal_vld_params
["ip-profile"]))
443 if internal_vld_params
.get("provider-network"):
445 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
446 internal_vld_params
["name"], "provider-network"),
447 internal_vld_params
["provider-network"].copy())
449 for icp_params
in get_iterable(internal_vld_params
, "internal-connection-point"):
452 for vdu_descriptor
in vnf_descriptor
["vdu"]:
453 for vdu_interface
in vdu_descriptor
["interface"]:
454 if vdu_interface
.get("internal-connection-point-ref") == icp_params
["id-ref"]:
455 if icp_params
.get("ip-address"):
456 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
457 vdu_descriptor
["id"], "interfaces",
458 vdu_interface
["name"], "ip_address"),
459 icp_params
["ip-address"])
461 if icp_params
.get("mac-address"):
462 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
463 vdu_descriptor
["id"], "interfaces",
464 vdu_interface
["name"], "mac_address"),
465 icp_params
["mac-address"])
471 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index[{}]:"
472 "internal-vld:id-ref={} is not present at vnfd:internal-"
473 "connection-point".format(vnf_params
["member-vnf-index"],
474 icp_params
["id-ref"]))
476 for vld_params
in get_iterable(ns_params
, "vld"):
477 if "ip-profile" in vld_params
:
478 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "ip-profile"),
479 ip_profile_2_RO(vld_params
["ip-profile"]))
481 if vld_params
.get("provider-network"):
483 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "provider-network"),
484 vld_params
["provider-network"].copy())
486 if "wimAccountId" in vld_params
and vld_params
["wimAccountId"] is not None:
487 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "wim_account"),
488 wim_account_2_RO(vld_params
["wimAccountId"])),
489 if vld_params
.get("vim-network-name"):
491 if isinstance(vld_params
["vim-network-name"], dict):
492 for vim_account
, vim_net
in vld_params
["vim-network-name"].items():
493 RO_vld_sites
.append({
494 "netmap-use": vim_net
,
495 "datacenter": vim_account_2_RO(vim_account
)
497 else: # isinstance str
498 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-name"]})
500 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
502 if vld_params
.get("vim-network-id"):
504 if isinstance(vld_params
["vim-network-id"], dict):
505 for vim_account
, vim_net
in vld_params
["vim-network-id"].items():
506 RO_vld_sites
.append({
507 "netmap-use": vim_net
,
508 "datacenter": vim_account_2_RO(vim_account
)
510 else: # isinstance str
511 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-id"]})
513 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
514 if vld_params
.get("ns-net"):
515 if isinstance(vld_params
["ns-net"], dict):
516 for vld_id
, instance_scenario_id
in vld_params
["ns-net"].items():
517 RO_vld_ns_net
= {"instance_scenario_id": instance_scenario_id
, "osm_id": vld_id
}
518 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "use-network"), RO_vld_ns_net
)
519 if "vnfd-connection-point-ref" in vld_params
:
520 for cp_params
in vld_params
["vnfd-connection-point-ref"]:
522 for constituent_vnfd
in nsd
["constituent-vnfd"]:
523 if constituent_vnfd
["member-vnf-index"] == cp_params
["member-vnf-index-ref"]:
524 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
528 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={} "
529 "is not present at nsd:constituent-vnfd".format(cp_params
["member-vnf-index-ref"]))
531 for vdu_descriptor
in vnf_descriptor
["vdu"]:
532 for interface_descriptor
in vdu_descriptor
["interface"]:
533 if interface_descriptor
.get("external-connection-point-ref") == \
534 cp_params
["vnfd-connection-point-ref"]:
541 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={}:"
542 "vnfd-connection-point-ref={} is not present at vnfd={}".format(
543 cp_params
["member-vnf-index-ref"],
544 cp_params
["vnfd-connection-point-ref"],
545 vnf_descriptor
["id"]))
546 if cp_params
.get("ip-address"):
547 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
548 vdu_descriptor
["id"], "interfaces",
549 interface_descriptor
["name"], "ip_address"),
550 cp_params
["ip-address"])
551 if cp_params
.get("mac-address"):
552 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
553 vdu_descriptor
["id"], "interfaces",
554 interface_descriptor
["name"], "mac_address"),
555 cp_params
["mac-address"])
558 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None):
559 # make a copy to do not change
560 vdu_create
= copy(vdu_create
)
561 vdu_delete
= copy(vdu_delete
)
563 vdurs
= db_vnfr
.get("vdur")
566 vdu_index
= len(vdurs
)
569 vdur
= vdurs
[vdu_index
]
570 if vdur
.get("pdu-type"):
572 vdu_id_ref
= vdur
["vdu-id-ref"]
573 if vdu_create
and vdu_create
.get(vdu_id_ref
):
574 for index
in range(0, vdu_create
[vdu_id_ref
]):
575 vdur
= deepcopy(vdur
)
576 vdur
["_id"] = str(uuid4())
577 vdur
["count-index"] += 1
578 vdurs
.insert(vdu_index
+1+index
, vdur
)
579 del vdu_create
[vdu_id_ref
]
580 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
582 vdu_delete
[vdu_id_ref
] -= 1
583 if not vdu_delete
[vdu_id_ref
]:
584 del vdu_delete
[vdu_id_ref
]
585 # check all operations are done
586 if vdu_create
or vdu_delete
:
587 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
590 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
593 vnfr_update
= {"vdur": vdurs
}
594 db_vnfr
["vdur"] = vdurs
595 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
597 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
599 Updates database nsr with the RO info for the created vld
600 :param ns_update_nsr: dictionary to be filled with the updated info
601 :param db_nsr: content of db_nsr. This is also modified
602 :param nsr_desc_RO: nsr descriptor from RO
603 :return: Nothing, LcmException is raised on errors
606 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
607 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
608 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
610 vld
["vim-id"] = net_RO
.get("vim_net_id")
611 vld
["name"] = net_RO
.get("vim_name")
612 vld
["status"] = net_RO
.get("status")
613 vld
["status-detailed"] = net_RO
.get("error_msg")
614 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
617 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
619 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
621 for db_vnfr
in db_vnfrs
.values():
622 vnfr_update
= {"status": "ERROR"}
623 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
624 if "status" not in vdur
:
625 vdur
["status"] = "ERROR"
626 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
628 vdur
["status-detailed"] = str(error_text
)
629 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
630 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
631 except DbException
as e
:
632 self
.logger
.error("Cannot update vnf. {}".format(e
))
634 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
636 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
637 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
638 :param nsr_desc_RO: nsr descriptor from RO
639 :return: Nothing, LcmException is raised on errors
641 for vnf_index
, db_vnfr
in db_vnfrs
.items():
642 for vnf_RO
in nsr_desc_RO
["vnfs"]:
643 if vnf_RO
["member_vnf_index"] != vnf_index
:
646 if vnf_RO
.get("ip_address"):
647 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
648 elif not db_vnfr
.get("ip-address"):
649 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
650 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
652 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
653 vdur_RO_count_index
= 0
654 if vdur
.get("pdu-type"):
656 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
657 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
659 if vdur
["count-index"] != vdur_RO_count_index
:
660 vdur_RO_count_index
+= 1
662 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
663 if vdur_RO
.get("ip_address"):
664 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
666 vdur
["ip-address"] = None
667 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
668 vdur
["name"] = vdur_RO
.get("vim_name")
669 vdur
["status"] = vdur_RO
.get("status")
670 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
671 for ifacer
in get_iterable(vdur
, "interfaces"):
672 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
673 if ifacer
["name"] == interface_RO
.get("internal_name"):
674 ifacer
["ip-address"] = interface_RO
.get("ip_address")
675 ifacer
["mac-address"] = interface_RO
.get("mac_address")
678 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
680 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
681 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
684 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
685 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
687 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
688 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
689 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
691 vld
["vim-id"] = net_RO
.get("vim_net_id")
692 vld
["name"] = net_RO
.get("vim_name")
693 vld
["status"] = net_RO
.get("status")
694 vld
["status-detailed"] = net_RO
.get("error_msg")
695 vnfr_update
["vld.{}".format(vld_index
)] = vld
698 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
699 vnf_index
, vld
["id"]))
701 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
705 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
707 def _get_ns_config_info(self
, nsr_id
):
709 Generates a mapping between vnf,vdu elements and the N2VC id
710 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
711 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
712 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
713 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
715 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
716 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
718 ns_config_info
= {"osm-config-mapping": mapping
}
719 for vca
in vca_deployed_list
:
720 if not vca
["member-vnf-index"]:
722 if not vca
["vdu_id"]:
723 mapping
[vca
["member-vnf-index"]] = vca
["application"]
725 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
727 return ns_config_info
730 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
):
732 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
733 primitives as verify-ssh-credentials, or config when needed
734 :param desc_primitive_list: information of the descriptor
735 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
736 this element contains a ssh public key
737 :return: The modified list. Can ba an empty list, but always a list
739 if desc_primitive_list
:
740 primitive_list
= desc_primitive_list
.copy()
743 # look for primitive config, and get the position. None if not present
744 config_position
= None
745 for index
, primitive
in enumerate(primitive_list
):
746 if primitive
["name"] == "config":
747 config_position
= index
750 # for NS, add always a config primitive if not present (bug 874)
751 if not vca_deployed
["member-vnf-index"] and config_position
is None:
752 primitive_list
.insert(0, {"name": "config", "parameter": []})
754 # for VNF/VDU add verify-ssh-credentials after config
755 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
756 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
757 return primitive_list
759 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
760 n2vc_key_list
, stage
):
763 :param logging_text: preffix text to use at logging
764 :param nsr_id: nsr identity
765 :param nsd: database content of ns descriptor
766 :param db_nsr: database content of ns record
767 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
769 :param db_vnfds_ref: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
770 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
771 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
772 :return: None or exception
776 RO_descriptor_number
= 0 # number of descriptors created at RO
777 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
778 nslcmop_id
= db_nslcmop
["_id"]
779 start_deploy
= time()
780 ns_params
= db_nslcmop
.get("operationParams")
781 if ns_params
and ns_params
.get("timeout_ns_deploy"):
782 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
784 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
786 # Check for and optionally request placement optimization. Database will be updated if placement activated
787 stage
[2] = "Waiting for Placement."
788 await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
)
792 # get vnfds, instantiate at RO
793 for c_vnf
in nsd
.get("constituent-vnfd", ()):
794 member_vnf_index
= c_vnf
["member-vnf-index"]
795 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
796 vnfd_ref
= vnfd
["id"]
798 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
799 db_nsr_update
["detailed-status"] = " ".join(stage
)
800 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
801 self
._write
_op
_status
(nslcmop_id
, stage
)
803 # self.logger.debug(logging_text + stage[2])
804 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
805 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
806 RO_descriptor_number
+= 1
808 # look position at deployed.RO.vnfd if not present it will be appended at the end
809 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
810 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
813 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
814 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
817 RO_update
= {"member-vnf-index": member_vnf_index
}
818 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
820 RO_update
["id"] = vnfd_list
[0]["uuid"]
821 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
822 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
824 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
825 get("additionalParamsForVnf"), nsr_id
)
826 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
827 RO_update
["id"] = desc
["uuid"]
828 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
829 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
830 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
831 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
836 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
837 db_nsr_update
["detailed-status"] = " ".join(stage
)
838 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
839 self
._write
_op
_status
(nslcmop_id
, stage
)
841 # self.logger.debug(logging_text + stage[2])
842 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
843 RO_descriptor_number
+= 1
844 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
846 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
847 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
848 nsd_ref
, RO_nsd_uuid
))
850 nsd_RO
= deepcopy(nsd
)
851 nsd_RO
["id"] = RO_osm_nsd_id
852 nsd_RO
.pop("_id", None)
853 nsd_RO
.pop("_admin", None)
854 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
855 member_vnf_index
= c_vnf
["member-vnf-index"]
856 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
857 for c_vld
in nsd_RO
.get("vld", ()):
858 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
859 member_vnf_index
= cp
["member-vnf-index-ref"]
860 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
862 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
863 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
864 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
865 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
866 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
869 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
870 db_nsr_update
["detailed-status"] = " ".join(stage
)
871 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
872 self
._write
_op
_status
(nslcmop_id
, stage
)
874 # if present use it unless in error status
875 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
878 stage
[2] = "Looking for existing ns at RO"
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] + " RO_ns_id={}".format(RO_nsr_id))
883 desc
= await self
.RO
.show("ns", RO_nsr_id
)
885 except ROclient
.ROClientException
as e
:
886 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
888 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
890 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
891 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
892 if ns_status
== "ERROR":
893 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
894 self
.logger
.debug(logging_text
+ stage
[2])
895 await self
.RO
.delete("ns", RO_nsr_id
)
896 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
898 stage
[2] = "Checking dependencies"
899 db_nsr_update
["detailed-status"] = " ".join(stage
)
900 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
901 self
._write
_op
_status
(nslcmop_id
, stage
)
902 # self.logger.debug(logging_text + stage[2])
904 # check if VIM is creating and wait look if previous tasks in process
905 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
907 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
908 self
.logger
.debug(logging_text
+ stage
[2])
909 await asyncio
.wait(task_dependency
, timeout
=3600)
910 if ns_params
.get("vnf"):
911 for vnf
in ns_params
["vnf"]:
912 if "vimAccountId" in vnf
:
913 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
916 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
917 self
.logger
.debug(logging_text
+ stage
[2])
918 await asyncio
.wait(task_dependency
, timeout
=3600)
920 stage
[2] = "Checking instantiation parameters."
921 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, db_vnfrs
, n2vc_key_list
)
922 stage
[2] = "Deploying ns at VIM."
923 db_nsr_update
["detailed-status"] = " ".join(stage
)
924 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
925 self
._write
_op
_status
(nslcmop_id
, stage
)
927 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
928 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
929 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
930 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
931 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
933 # wait until NS is ready
934 stage
[2] = "Waiting VIM to deploy ns."
935 db_nsr_update
["detailed-status"] = " ".join(stage
)
936 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
937 self
._write
_op
_status
(nslcmop_id
, stage
)
938 detailed_status_old
= None
939 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
942 while time() <= start_deploy
+ timeout_ns_deploy
:
943 desc
= await self
.RO
.show("ns", RO_nsr_id
)
947 # desc has changed => update db
948 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
951 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
952 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
953 if ns_status
== "ERROR":
954 raise ROclient
.ROClientException(ns_status_info
)
955 elif ns_status
== "BUILD":
956 stage
[2] = "VIM: ({})".format(ns_status_info
)
957 elif ns_status
== "ACTIVE":
958 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
960 self
.ns_update_vnfr(db_vnfrs
, desc
)
962 except LcmExceptionNoMgmtIP
:
965 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
966 if stage
[2] != detailed_status_old
:
967 detailed_status_old
= stage
[2]
968 db_nsr_update
["detailed-status"] = " ".join(stage
)
969 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
970 self
._write
_op
_status
(nslcmop_id
, stage
)
971 await asyncio
.sleep(5, loop
=self
.loop
)
972 else: # timeout_ns_deploy
973 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
976 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
978 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
979 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
980 stage
[2] = "Deployed at VIM"
981 db_nsr_update
["detailed-status"] = " ".join(stage
)
982 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
983 self
._write
_op
_status
(nslcmop_id
, stage
)
984 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
985 # self.logger.debug(logging_text + "Deployed at VIM")
986 except (ROclient
.ROClientException
, LcmException
, DbException
) as e
:
987 stage
[2] = "ERROR deploying at VIM"
988 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
991 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
993 Wait for ip addres at RO, and optionally, insert public key in virtual machine
994 :param logging_text: prefix use for logging
999 :param pub_key: public ssh key to inject, None to skip
1000 :param user: user to apply the public ssh key
1004 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
1008 target_vdu_id
= None
1014 if ro_retries
>= 360: # 1 hour
1015 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
1017 await asyncio
.sleep(10, loop
=self
.loop
)
1020 if not target_vdu_id
:
1021 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1023 if not vdu_id
: # for the VNF case
1024 if db_vnfr
.get("status") == "ERROR":
1025 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1026 ip_address
= db_vnfr
.get("ip-address")
1029 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1031 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1032 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1034 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1035 vdur
= db_vnfr
["vdur"][0]
1037 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1040 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1041 ip_address
= vdur
.get("ip-address")
1044 target_vdu_id
= vdur
["vdu-id-ref"]
1045 elif vdur
.get("status") == "ERROR":
1046 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1048 if not target_vdu_id
:
1051 # inject public key into machine
1052 if pub_key
and user
:
1053 # wait until NS is deployed at RO
1055 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1056 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1060 # self.logger.debug(logging_text + "Inserting RO key")
1061 if vdur
.get("pdu-type"):
1062 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1065 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1066 result_dict
= await self
.RO
.create_action(
1068 item_id_name
=ro_nsr_id
,
1069 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1071 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1072 if not result_dict
or not isinstance(result_dict
, dict):
1073 raise LcmException("Unknown response from RO when injecting key")
1074 for result
in result_dict
.values():
1075 if result
.get("vim_result") == 200:
1078 raise ROclient
.ROClientException("error injecting key: {}".format(
1079 result
.get("description")))
1081 except ROclient
.ROClientException
as e
:
1083 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1087 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1093 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1095 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1097 my_vca
= vca_deployed_list
[vca_index
]
1098 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1099 # vdu or kdu: no dependencies
1103 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1104 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1105 configuration_status_list
= db_nsr
["configurationStatus"]
1106 for index
, vca_deployed
in enumerate(configuration_status_list
):
1107 if index
== vca_index
:
1110 if not my_vca
.get("member-vnf-index") or \
1111 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1112 internal_status
= configuration_status_list
[index
].get("status")
1113 if internal_status
== 'READY':
1115 elif internal_status
== 'BROKEN':
1116 raise LcmException("Configuration aborted because dependent charm/s has failed")
1120 # no dependencies, return
1122 await asyncio
.sleep(10)
1125 raise LcmException("Configuration aborted because dependent charm/s timeout")
1127 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1128 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
):
1129 nsr_id
= db_nsr
["_id"]
1130 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1131 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1132 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1134 'collection': 'nsrs',
1135 'filter': {'_id': nsr_id
},
1136 'path': db_update_entry
1142 element_under_configuration
= nsr_id
1146 vnfr_id
= db_vnfr
["_id"]
1148 namespace
= "{nsi}.{ns}".format(
1149 nsi
=nsi_id
if nsi_id
else "",
1153 element_type
= 'VNF'
1154 element_under_configuration
= vnfr_id
1155 namespace
+= ".{}".format(vnfr_id
)
1157 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1158 element_type
= 'VDU'
1159 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1161 namespace
+= ".{}".format(kdu_name
)
1162 element_type
= 'KDU'
1163 element_under_configuration
= kdu_name
1166 self
.fs
.sync() # Sync from FSMongo
1167 artifact_path
= "{}/{}/charms/{}".format(
1168 base_folder
["folder"],
1169 base_folder
["pkg-dir"],
1170 config_descriptor
["juju"]["charm"]
1173 is_proxy_charm
= deep_get(config_descriptor
, ('juju', 'charm')) is not None
1174 if deep_get(config_descriptor
, ('juju', 'proxy')) is False:
1175 is_proxy_charm
= False
1177 # n2vc_redesign STEP 3.1
1179 # find old ee_id if exists
1180 ee_id
= vca_deployed
.get("ee_id")
1182 # create or register execution environment in VCA
1185 self
._write
_configuration
_status
(
1187 vca_index
=vca_index
,
1189 element_under_configuration
=element_under_configuration
,
1190 element_type
=element_type
1193 step
= "create execution environment"
1194 self
.logger
.debug(logging_text
+ step
)
1195 ee_id
, credentials
= await self
.n2vc
.create_execution_environment(namespace
=namespace
,
1200 step
= "Waiting to VM being up and getting IP address"
1201 self
.logger
.debug(logging_text
+ step
)
1202 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1203 user
=None, pub_key
=None)
1204 credentials
= {"hostname": rw_mgmt_ip
}
1206 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1207 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1208 # merged. Meanwhile let's get username from initial-config-primitive
1209 if not username
and config_descriptor
.get("initial-config-primitive"):
1210 for config_primitive
in config_descriptor
["initial-config-primitive"]:
1211 for param
in config_primitive
.get("parameter", ()):
1212 if param
["name"] == "ssh-username":
1213 username
= param
["value"]
1216 raise LcmException("Cannot determine the username neither with 'initial-config-promitive' nor with "
1217 "'config-access.ssh-access.default-user'")
1218 credentials
["username"] = username
1219 # n2vc_redesign STEP 3.2
1221 self
._write
_configuration
_status
(
1223 vca_index
=vca_index
,
1224 status
='REGISTERING',
1225 element_under_configuration
=element_under_configuration
,
1226 element_type
=element_type
1229 step
= "register execution environment {}".format(credentials
)
1230 self
.logger
.debug(logging_text
+ step
)
1231 ee_id
= await self
.n2vc
.register_execution_environment(credentials
=credentials
, namespace
=namespace
,
1234 # for compatibility with MON/POL modules, the need model and application name at database
1235 # TODO ask to N2VC instead of assuming the format "model_name.application_name"
1236 ee_id_parts
= ee_id
.split('.')
1237 model_name
= ee_id_parts
[0]
1238 application_name
= ee_id_parts
[1]
1239 db_nsr_update
= {db_update_entry
+ "model": model_name
,
1240 db_update_entry
+ "application": application_name
,
1241 db_update_entry
+ "ee_id": ee_id
}
1243 # n2vc_redesign STEP 3.3
1245 step
= "Install configuration Software"
1247 self
._write
_configuration
_status
(
1249 vca_index
=vca_index
,
1250 status
='INSTALLING SW',
1251 element_under_configuration
=element_under_configuration
,
1252 element_type
=element_type
,
1253 other_update
=db_nsr_update
1256 # TODO check if already done
1257 self
.logger
.debug(logging_text
+ step
)
1259 if not is_proxy_charm
:
1260 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1261 if initial_config_primitive_list
:
1262 for primitive
in initial_config_primitive_list
:
1263 if primitive
["name"] == "config":
1264 config
= self
._map
_primitive
_params
(
1270 await self
.n2vc
.install_configuration_sw(
1272 artifact_path
=artifact_path
,
1277 # write in db flag of configuration_sw already installed
1278 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1280 # add relations for this VCA (wait for other peers related with this VCA)
1281 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
, vca_index
=vca_index
)
1283 # if SSH access is required, then get execution environment SSH public
1284 if is_proxy_charm
: # if native charm we have waited already to VM be UP
1287 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1288 # Needed to inject a ssh key
1289 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1290 step
= "Install configuration Software, getting public ssh key"
1291 pub_key
= await self
.n2vc
.get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1293 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1295 step
= "Waiting to VM being up and getting IP address"
1296 self
.logger
.debug(logging_text
+ step
)
1298 # n2vc_redesign STEP 5.1
1299 # wait for RO (ip-address) Insert pub_key into VM
1301 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1302 user
=user
, pub_key
=pub_key
)
1304 rw_mgmt_ip
= None # This is for a NS configuration
1306 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1308 # store rw_mgmt_ip in deploy params for later replacement
1309 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1311 # n2vc_redesign STEP 6 Execute initial config primitive
1312 step
= 'execute initial config primitive'
1313 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1315 # sort initial config primitives by 'seq'
1316 if initial_config_primitive_list
:
1318 initial_config_primitive_list
.sort(key
=lambda val
: int(val
['seq']))
1319 except Exception as e
:
1320 self
.logger
.error(logging_text
+ step
+ ": " + str(e
))
1322 self
.logger
.debug(logging_text
+ step
+ ": No initial-config-primitive")
1324 # add config if not present for NS charm
1325 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1328 # wait for dependent primitives execution (NS -> VNF -> VDU)
1329 if initial_config_primitive_list
:
1330 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1332 # stage, in function of element type: vdu, kdu, vnf or ns
1333 my_vca
= vca_deployed_list
[vca_index
]
1334 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1336 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1337 elif my_vca
.get("member-vnf-index"):
1339 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1342 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1344 self
._write
_configuration
_status
(
1346 vca_index
=vca_index
,
1347 status
='EXECUTING PRIMITIVE'
1350 self
._write
_op
_status
(
1355 check_if_terminated_needed
= True
1356 for initial_config_primitive
in initial_config_primitive_list
:
1357 # adding information on the vca_deployed if it is a NS execution environment
1358 if not vca_deployed
["member-vnf-index"]:
1359 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1360 # TODO check if already done
1361 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1363 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1364 self
.logger
.debug(logging_text
+ step
)
1365 await self
.n2vc
.exec_primitive(
1367 primitive_name
=initial_config_primitive
["name"],
1368 params_dict
=primitive_params_
,
1371 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1372 if check_if_terminated_needed
:
1373 if config_descriptor
.get('terminate-config-primitive'):
1374 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1375 check_if_terminated_needed
= False
1377 # TODO register in database that primitive is done
1379 step
= "instantiated at VCA"
1380 self
.logger
.debug(logging_text
+ step
)
1382 self
._write
_configuration
_status
(
1384 vca_index
=vca_index
,
1388 except Exception as e
: # TODO not use Exception but N2VC exception
1389 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1390 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1391 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1392 self
._write
_configuration
_status
(
1394 vca_index
=vca_index
,
1397 raise LcmException("{} {}".format(step
, e
)) from e
1399 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1400 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1402 Update db_nsr fields.
1405 :param current_operation:
1406 :param current_operation_id:
1407 :param error_description:
1408 :param error_detail:
1409 :param other_update: Other required changes at database if provided, will be cleared
1413 db_dict
= other_update
or {}
1414 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1415 db_dict
["_admin.current-operation"] = current_operation_id
1416 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1417 db_dict
["currentOperation"] = current_operation
1418 db_dict
["currentOperationID"] = current_operation_id
1419 db_dict
["errorDescription"] = error_description
1420 db_dict
["errorDetail"] = error_detail
1423 db_dict
["nsState"] = ns_state
1424 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1425 except DbException
as e
:
1426 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1428 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1429 operation_state
: str = None, other_update
: dict = None):
1431 db_dict
= other_update
or {}
1432 db_dict
['queuePosition'] = queuePosition
1433 if isinstance(stage
, list):
1434 db_dict
['stage'] = stage
[0]
1435 db_dict
['detailed-status'] = " ".join(stage
)
1436 elif stage
is not None:
1437 db_dict
['stage'] = str(stage
)
1439 if error_message
is not None:
1440 db_dict
['errorMessage'] = error_message
1441 if operation_state
is not None:
1442 db_dict
['operationState'] = operation_state
1443 db_dict
["statusEnteredTime"] = time()
1444 self
.update_db_2("nslcmops", op_id
, db_dict
)
1445 except DbException
as e
:
1446 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1448 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1450 nsr_id
= db_nsr
["_id"]
1451 # configurationStatus
1452 config_status
= db_nsr
.get('configurationStatus')
1454 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1455 enumerate(config_status
) if v
}
1457 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1459 except DbException
as e
:
1460 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1462 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1463 element_under_configuration
: str = None, element_type
: str = None,
1464 other_update
: dict = None):
1466 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1467 # .format(vca_index, status))
1470 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1471 db_dict
= other_update
or {}
1473 db_dict
[db_path
+ 'status'] = status
1474 if element_under_configuration
:
1475 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1477 db_dict
[db_path
+ 'elementType'] = element_type
1478 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1479 except DbException
as e
:
1480 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1481 .format(status
, nsr_id
, vca_index
, e
))
1483 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1485 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1486 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1487 Database is used because the result can be obtained from a different LCM worker in case of HA.
1488 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1489 :param db_nslcmop: database content of nslcmop
1490 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1491 :return: None. Modifies database vnfrs and parameter db_vnfr with the computed 'vim-account-id'
1493 nslcmop_id
= db_nslcmop
['_id']
1494 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1495 if placement_engine
== "PLA":
1496 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1497 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1498 db_poll_interval
= 5
1499 wait
= db_poll_interval
* 10
1501 while not pla_result
and wait
>= 0:
1502 await asyncio
.sleep(db_poll_interval
)
1503 wait
-= db_poll_interval
1504 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1505 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1508 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1510 for pla_vnf
in pla_result
['vnf']:
1511 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1512 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1514 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1516 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1519 def update_nsrs_with_pla_result(self
, params
):
1521 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1522 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1523 except Exception as e
:
1524 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1526 async def instantiate(self
, nsr_id
, nslcmop_id
):
1529 :param nsr_id: ns instance to deploy
1530 :param nslcmop_id: operation to run
1534 # Try to lock HA task here
1535 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1536 if not task_is_locked_by_me
:
1537 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1540 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1541 self
.logger
.debug(logging_text
+ "Enter")
1543 # get all needed from database
1545 # database nsrs record
1548 # database nslcmops record
1551 # update operation on nsrs
1553 # update operation on nslcmops
1554 db_nslcmop_update
= {}
1556 nslcmop_operation_state
= None
1557 db_vnfrs
= {} # vnf's info indexed by member-index
1559 tasks_dict_info
= {} # from task to info text
1562 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1563 # ^ stage, step, VIM progress
1565 # wait for any previous tasks in process
1566 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1568 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1569 stage
[1] = "Reading from database,"
1570 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1571 db_nsr_update
["detailed-status"] = "creating"
1572 db_nsr_update
["operational-status"] = "init"
1573 self
._write
_ns
_status
(
1575 ns_state
="BUILDING",
1576 current_operation
="INSTANTIATING",
1577 current_operation_id
=nslcmop_id
,
1578 other_update
=db_nsr_update
1580 self
._write
_op
_status
(
1586 # read from db: operation
1587 stage
[1] = "Getting nslcmop={} from db".format(nslcmop_id
)
1588 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1589 ns_params
= db_nslcmop
.get("operationParams")
1590 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1591 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1593 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1596 stage
[1] = "Getting nsr={} from db".format(nsr_id
)
1597 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1598 # nsd is replicated into ns (no db read)
1600 # nsr_name = db_nsr["name"] # TODO short-name??
1602 # read from db: vnf's of this ns
1603 stage
[1] = "Getting vnfrs from db"
1604 self
.logger
.debug(logging_text
+ stage
[1])
1605 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1607 # read from db: vnfd's for every vnf
1608 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1609 db_vnfds
= {} # every vnfd data indexed by vnf id
1610 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1612 # for each vnf in ns, read vnfd
1613 for vnfr
in db_vnfrs_list
:
1614 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1615 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1616 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1617 # if we haven't this vnfd, read it from db
1618 if vnfd_id
not in db_vnfds
:
1620 stage
[1] = "Getting vnfd={} id='{}' from db".format(vnfd_id
, vnfd_ref
)
1621 self
.logger
.debug(logging_text
+ stage
[1])
1622 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1625 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1626 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1627 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1629 # Get or generates the _admin.deployed.VCA list
1630 vca_deployed_list
= None
1631 if db_nsr
["_admin"].get("deployed"):
1632 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1633 if vca_deployed_list
is None:
1634 vca_deployed_list
= []
1635 configuration_status_list
= []
1636 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1637 db_nsr_update
["configurationStatus"] = configuration_status_list
1638 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1639 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1640 elif isinstance(vca_deployed_list
, dict):
1641 # maintain backward compatibility. Change a dict to list at database
1642 vca_deployed_list
= list(vca_deployed_list
.values())
1643 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1644 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1646 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1647 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1648 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1650 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1651 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1652 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1654 # n2vc_redesign STEP 2 Deploy Network Scenario
1655 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1656 self
._write
_op
_status
(
1661 stage
[1] = "Deploying KDUs,"
1662 # self.logger.debug(logging_text + "Before deploy_kdus")
1663 # Call to deploy_kdus in case exists the "vdu:kdu" param
1664 await self
.deploy_kdus(
1665 logging_text
=logging_text
,
1667 nslcmop_id
=nslcmop_id
,
1670 task_instantiation_info
=tasks_dict_info
,
1673 stage
[1] = "Getting VCA public key."
1674 # n2vc_redesign STEP 1 Get VCA public ssh-key
1675 # feature 1429. Add n2vc public key to needed VMs
1676 n2vc_key
= self
.n2vc
.get_public_key()
1677 n2vc_key_list
= [n2vc_key
]
1678 if self
.vca_config
.get("public_key"):
1679 n2vc_key_list
.append(self
.vca_config
["public_key"])
1681 stage
[1] = "Deploying NS at VIM."
1682 task_ro
= asyncio
.ensure_future(
1683 self
.instantiate_RO(
1684 logging_text
=logging_text
,
1688 db_nslcmop
=db_nslcmop
,
1690 db_vnfds_ref
=db_vnfds_ref
,
1691 n2vc_key_list
=n2vc_key_list
,
1695 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
1696 tasks_dict_info
[task_ro
] = "Deploying at VIM"
1698 # n2vc_redesign STEP 3 to 6 Deploy N2VC
1699 stage
[1] = "Deploying Execution Environments."
1700 self
.logger
.debug(logging_text
+ stage
[1])
1702 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
1703 # get_iterable() returns a value from a dict or empty tuple if key does not exist
1704 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
1705 vnfd_id
= c_vnf
["vnfd-id-ref"]
1706 vnfd
= db_vnfds_ref
[vnfd_id
]
1707 member_vnf_index
= str(c_vnf
["member-vnf-index"])
1708 db_vnfr
= db_vnfrs
[member_vnf_index
]
1709 base_folder
= vnfd
["_admin"]["storage"]
1715 # Get additional parameters
1717 if db_vnfr
.get("additionalParamsForVnf"):
1718 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
1720 descriptor_config
= vnfd
.get("vnf-configuration")
1721 if descriptor_config
and descriptor_config
.get("juju"):
1723 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
1726 nslcmop_id
=nslcmop_id
,
1732 member_vnf_index
=member_vnf_index
,
1733 vdu_index
=vdu_index
,
1735 deploy_params
=deploy_params
,
1736 descriptor_config
=descriptor_config
,
1737 base_folder
=base_folder
,
1738 task_instantiation_info
=tasks_dict_info
,
1742 # Deploy charms for each VDU that supports one.
1743 for vdud
in get_iterable(vnfd
, 'vdu'):
1745 descriptor_config
= vdud
.get('vdu-configuration')
1746 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
1747 if vdur
.get("additionalParams"):
1748 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
1750 deploy_params_vdu
= deploy_params
1751 if descriptor_config
and descriptor_config
.get("juju"):
1752 # look for vdu index in the db_vnfr["vdu"] section
1753 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1754 # if vdur["vdu-id-ref"] == vdu_id:
1757 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1758 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1759 # vdu_name = vdur.get("name")
1762 for vdu_index
in range(int(vdud
.get("count", 1))):
1763 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
1765 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
1766 member_vnf_index
, vdu_id
, vdu_index
),
1769 nslcmop_id
=nslcmop_id
,
1775 member_vnf_index
=member_vnf_index
,
1776 vdu_index
=vdu_index
,
1778 deploy_params
=deploy_params_vdu
,
1779 descriptor_config
=descriptor_config
,
1780 base_folder
=base_folder
,
1781 task_instantiation_info
=tasks_dict_info
,
1784 for kdud
in get_iterable(vnfd
, 'kdu'):
1785 kdu_name
= kdud
["name"]
1786 descriptor_config
= kdud
.get('kdu-configuration')
1787 if descriptor_config
and descriptor_config
.get("juju"):
1791 # look for vdu index in the db_vnfr["vdu"] section
1792 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1793 # if vdur["vdu-id-ref"] == vdu_id:
1796 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1797 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1798 # vdu_name = vdur.get("name")
1802 logging_text
=logging_text
,
1805 nslcmop_id
=nslcmop_id
,
1811 member_vnf_index
=member_vnf_index
,
1812 vdu_index
=vdu_index
,
1814 deploy_params
=deploy_params
,
1815 descriptor_config
=descriptor_config
,
1816 base_folder
=base_folder
,
1817 task_instantiation_info
=tasks_dict_info
,
1821 # Check if this NS has a charm configuration
1822 descriptor_config
= nsd
.get("ns-configuration")
1823 if descriptor_config
and descriptor_config
.get("juju"):
1826 member_vnf_index
= None
1832 # Get additional parameters
1834 if db_nsr
.get("additionalParamsForNs"):
1835 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
1836 base_folder
= nsd
["_admin"]["storage"]
1838 logging_text
=logging_text
,
1841 nslcmop_id
=nslcmop_id
,
1847 member_vnf_index
=member_vnf_index
,
1848 vdu_index
=vdu_index
,
1850 deploy_params
=deploy_params
,
1851 descriptor_config
=descriptor_config
,
1852 base_folder
=base_folder
,
1853 task_instantiation_info
=tasks_dict_info
,
1857 # rest of staff will be done at finally
1859 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
1860 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
1862 except asyncio
.CancelledError
:
1863 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
1864 exc
= "Operation was cancelled"
1865 except Exception as e
:
1866 exc
= traceback
.format_exc()
1867 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
1870 error_list
.append(str(exc
))
1872 # wait for pending tasks
1874 stage
[1] = "Waiting for instantiate pending tasks."
1875 self
.logger
.debug(logging_text
+ stage
[1])
1876 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
1877 stage
, nslcmop_id
, nsr_id
=nsr_id
)
1878 stage
[1] = stage
[2] = ""
1879 except asyncio
.CancelledError
:
1880 error_list
.append("Cancelled")
1881 # TODO cancel all tasks
1882 except Exception as exc
:
1883 error_list
.append(str(exc
))
1885 # update operation-status
1886 db_nsr_update
["operational-status"] = "running"
1887 # let's begin with VCA 'configured' status (later we can change it)
1888 db_nsr_update
["config-status"] = "configured"
1889 for task
, task_name
in tasks_dict_info
.items():
1890 if not task
.done() or task
.cancelled() or task
.exception():
1891 if task_name
.startswith(self
.task_name_deploy_vca
):
1892 # A N2VC task is pending
1893 db_nsr_update
["config-status"] = "failed"
1895 # RO or KDU task is pending
1896 db_nsr_update
["operational-status"] = "failed"
1898 # update status at database
1900 error_detail
= ". ".join(error_list
)
1901 self
.logger
.error(logging_text
+ error_detail
)
1902 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
1903 error_description_nsr
= 'Operation: INSTANTIATING.{}, Stage {}'.format(nslcmop_id
, stage
[0])
1905 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
1906 db_nslcmop_update
["detailed-status"] = error_detail
1907 nslcmop_operation_state
= "FAILED"
1911 error_description_nsr
= error_description_nslcmop
= None
1913 db_nsr_update
["detailed-status"] = "Done"
1914 db_nslcmop_update
["detailed-status"] = "Done"
1915 nslcmop_operation_state
= "COMPLETED"
1918 self
._write
_ns
_status
(
1921 current_operation
="IDLE",
1922 current_operation_id
=None,
1923 error_description
=error_description_nsr
,
1924 error_detail
=error_detail
,
1925 other_update
=db_nsr_update
1928 self
._write
_op
_status
(
1931 error_message
=error_description_nslcmop
,
1932 operation_state
=nslcmop_operation_state
,
1933 other_update
=db_nslcmop_update
,
1936 if nslcmop_operation_state
:
1938 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
1939 "operationState": nslcmop_operation_state
},
1941 except Exception as e
:
1942 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
1944 self
.logger
.debug(logging_text
+ "Exit")
1945 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
1947 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int, timeout
: int = 3600) -> bool:
1950 # 1. find all relations for this VCA
1951 # 2. wait for other peers related
1956 # STEP 1: find all relations for this VCA
1959 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1962 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
1964 # read all ns-configuration relations
1965 ns_relations
= list()
1966 db_ns_relations
= deep_get(db_nsr
, ('nsd', 'ns-configuration', 'relation'))
1968 for r
in db_ns_relations
:
1969 # check if this VCA is in the relation
1970 if my_vca
.get('member-vnf-index') in\
1971 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1972 ns_relations
.append(r
)
1974 # read all vnf-configuration relations
1975 vnf_relations
= list()
1976 db_vnfd_list
= db_nsr
.get('vnfd-id')
1978 for vnfd
in db_vnfd_list
:
1979 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
1980 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
1981 if db_vnf_relations
:
1982 for r
in db_vnf_relations
:
1983 # check if this VCA is in the relation
1984 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1985 vnf_relations
.append(r
)
1987 # if no relations, terminate
1988 if not ns_relations
and not vnf_relations
:
1989 self
.logger
.debug(logging_text
+ ' No relations')
1992 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
1999 if now
- start
>= timeout
:
2000 self
.logger
.error(logging_text
+ ' : timeout adding relations')
2003 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2004 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2006 # for each defined NS relation, find the VCA's related
2007 for r
in ns_relations
:
2008 from_vca_ee_id
= None
2010 from_vca_endpoint
= None
2011 to_vca_endpoint
= None
2012 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2013 for vca
in vca_list
:
2014 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
2015 and vca
.get('config_sw_installed'):
2016 from_vca_ee_id
= vca
.get('ee_id')
2017 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2018 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2019 and vca
.get('config_sw_installed'):
2020 to_vca_ee_id
= vca
.get('ee_id')
2021 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2022 if from_vca_ee_id
and to_vca_ee_id
:
2024 await self
.n2vc
.add_relation(
2025 ee_id_1
=from_vca_ee_id
,
2026 ee_id_2
=to_vca_ee_id
,
2027 endpoint_1
=from_vca_endpoint
,
2028 endpoint_2
=to_vca_endpoint
)
2029 # remove entry from relations list
2030 ns_relations
.remove(r
)
2032 # check failed peers
2034 vca_status_list
= db_nsr
.get('configurationStatus')
2036 for i
in range(len(vca_list
)):
2038 vca_status
= vca_status_list
[i
]
2039 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2040 if vca_status
.get('status') == 'BROKEN':
2041 # peer broken: remove relation from list
2042 ns_relations
.remove(r
)
2043 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2044 if vca_status
.get('status') == 'BROKEN':
2045 # peer broken: remove relation from list
2046 ns_relations
.remove(r
)
2051 # for each defined VNF relation, find the VCA's related
2052 for r
in vnf_relations
:
2053 from_vca_ee_id
= None
2055 from_vca_endpoint
= None
2056 to_vca_endpoint
= None
2057 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2058 for vca
in vca_list
:
2059 if vca
.get('vdu_id') == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2060 from_vca_ee_id
= vca
.get('ee_id')
2061 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2062 if vca
.get('vdu_id') == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2063 to_vca_ee_id
= vca
.get('ee_id')
2064 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2065 if from_vca_ee_id
and to_vca_ee_id
:
2067 await self
.n2vc
.add_relation(
2068 ee_id_1
=from_vca_ee_id
,
2069 ee_id_2
=to_vca_ee_id
,
2070 endpoint_1
=from_vca_endpoint
,
2071 endpoint_2
=to_vca_endpoint
)
2072 # remove entry from relations list
2073 vnf_relations
.remove(r
)
2075 # check failed peers
2077 vca_status_list
= db_nsr
.get('configurationStatus')
2079 for i
in range(len(vca_list
)):
2081 vca_status
= vca_status_list
[i
]
2082 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2083 if vca_status
.get('status') == 'BROKEN':
2084 # peer broken: remove relation from list
2085 ns_relations
.remove(r
)
2086 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2087 if vca_status
.get('status') == 'BROKEN':
2088 # peer broken: remove relation from list
2089 ns_relations
.remove(r
)
2095 await asyncio
.sleep(5.0)
2097 if not ns_relations
and not vnf_relations
:
2098 self
.logger
.debug('Relations added')
2103 except Exception as e
:
2104 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2107 def _write_db_callback(self
, task
, item
, _id
, on_done
=None, on_exc
=None):
2109 callback for kdu install intended to store the returned kdu_instance at database
2114 result
= task
.result()
2116 db_update
[on_done
] = str(result
)
2117 except Exception as e
:
2119 db_update
[on_exc
] = str(e
)
2122 self
.update_db_2(item
, _id
, db_update
)
2126 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2127 # Launch kdus if present in the descriptor
2129 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2131 def _get_cluster_id(cluster_id
, cluster_type
):
2132 nonlocal k8scluster_id_2_uuic
2133 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2134 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2136 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2137 if not db_k8scluster
:
2138 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2139 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2141 raise LcmException("K8s cluster '{}' has not been initilized for '{}'".format(cluster_id
, cluster_type
))
2142 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2145 logging_text
+= "Deploy kdus: "
2148 db_nsr_update
= {"_admin.deployed.K8s": []}
2149 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2152 updated_cluster_list
= []
2154 for vnfr_data
in db_vnfrs
.values():
2155 for kdur
in get_iterable(vnfr_data
, "kdur"):
2156 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2157 vnfd_id
= vnfr_data
.get('vnfd-id')
2158 namespace
= kdur
.get("k8s-namespace")
2159 if kdur
.get("helm-chart"):
2160 kdumodel
= kdur
["helm-chart"]
2161 k8sclustertype
= "helm-chart"
2162 elif kdur
.get("juju-bundle"):
2163 kdumodel
= kdur
["juju-bundle"]
2164 k8sclustertype
= "juju-bundle"
2166 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2167 "juju-bundle. Maybe an old NBI version is running".
2168 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2169 # check if kdumodel is a file and exists
2171 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2172 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2173 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2174 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["'pkg-dir"], k8sclustertype
,
2176 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2177 kdumodel
= self
.fs
.path
+ filename
2178 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2180 except Exception: # it is not a file
2183 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2184 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2185 cluster_uuid
= _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2187 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2188 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2189 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2190 if del_repo_list
or added_repo_dict
:
2191 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2192 updated
= {'_admin.helm_charts_added.' +
2193 item
: name
for item
, name
in added_repo_dict
.items()}
2194 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2195 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2197 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2198 updated_cluster_list
.append(cluster_uuid
)
2200 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2201 kdur
["kdu-name"], k8s_cluster_id
)
2203 k8s_instace_info
= {"kdu-instance": None,
2204 "k8scluster-uuid": cluster_uuid
,
2205 "k8scluster-type": k8sclustertype
,
2206 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2207 "kdu-name": kdur
["kdu-name"],
2208 "kdu-model": kdumodel
,
2209 "namespace": namespace
}
2210 db_path
= "_admin.deployed.K8s.{}".format(index
)
2211 db_nsr_update
[db_path
] = k8s_instace_info
2212 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2214 db_dict
= {"collection": "nsrs",
2215 "filter": {"_id": nsr_id
},
2218 task
= asyncio
.ensure_future(
2219 self
.k8scluster_map
[k8sclustertype
].install(cluster_uuid
=cluster_uuid
, kdu_model
=kdumodel
,
2220 atomic
=True, params
=desc_params
,
2221 db_dict
=db_dict
, timeout
=600,
2222 kdu_name
=kdur
["kdu-name"], namespace
=namespace
))
2224 task
.add_done_callback(partial(self
._write
_db
_callback
, item
="nsrs", _id
=nsr_id
,
2225 on_done
=db_path
+ ".kdu-instance",
2226 on_exc
=db_path
+ ".detailed-status"))
2227 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2228 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2232 except (LcmException
, asyncio
.CancelledError
):
2234 except Exception as e
:
2235 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2236 if isinstance(e
, (N2VCException
, DbException
)):
2237 self
.logger
.error(logging_text
+ msg
)
2239 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2240 raise LcmException(msg
)
2243 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2245 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2246 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2247 base_folder
, task_instantiation_info
, stage
):
2248 # launch instantiate_N2VC in a asyncio task and register task object
2249 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2250 # if not found, create one entry and update database
2252 # fill db_nsr._admin.deployed.VCA.<index>
2254 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2255 if not vca_deployed
:
2257 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2258 vca_deployed
.get("vdu_id") == vdu_id
and \
2259 vca_deployed
.get("kdu_name") == kdu_name
and \
2260 vca_deployed
.get("vdu_count_index", 0) == vdu_index
:
2263 # not found, create one.
2265 "member-vnf-index": member_vnf_index
,
2267 "kdu_name": kdu_name
,
2268 "vdu_count_index": vdu_index
,
2269 "operational-status": "init", # TODO revise
2270 "detailed-status": "", # TODO revise
2271 "step": "initial-deploy", # TODO revise
2273 "vdu_name": vdu_name
,
2277 # create VCA and configurationStatus in db
2279 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2280 "configurationStatus.{}".format(vca_index
): dict()
2282 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2284 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2287 task_n2vc
= asyncio
.ensure_future(
2288 self
.instantiate_N2VC(
2289 logging_text
=logging_text
,
2290 vca_index
=vca_index
,
2296 vdu_index
=vdu_index
,
2297 deploy_params
=deploy_params
,
2298 config_descriptor
=descriptor_config
,
2299 base_folder
=base_folder
,
2300 nslcmop_id
=nslcmop_id
,
2304 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2305 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2306 member_vnf_index
or "", vdu_id
or "")
2308 # Check if this VNFD has a configured terminate action
2309 def _has_terminate_config_primitive(self
, vnfd
):
2310 vnf_config
= vnfd
.get("vnf-configuration")
2311 if vnf_config
and vnf_config
.get("terminate-config-primitive"):
2317 def _get_terminate_config_primitive_seq_list(vnfd
):
2318 """ Get a numerically sorted list of the sequences for this VNFD's terminate action """
2319 # No need to check for existing primitive twice, already done before
2320 vnf_config
= vnfd
.get("vnf-configuration")
2321 seq_list
= vnf_config
.get("terminate-config-primitive")
2322 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2323 seq_list_sorted
= sorted(seq_list
, key
=lambda x
: int(x
['seq']))
2324 return seq_list_sorted
2327 def _create_nslcmop(nsr_id
, operation
, params
):
2329 Creates a ns-lcm-opp content to be stored at database.
2330 :param nsr_id: internal id of the instance
2331 :param operation: instantiate, terminate, scale, action, ...
2332 :param params: user parameters for the operation
2333 :return: dictionary following SOL005 format
2335 # Raise exception if invalid arguments
2336 if not (nsr_id
and operation
and params
):
2338 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2344 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2345 "operationState": "PROCESSING",
2346 "statusEnteredTime": now
,
2347 "nsInstanceId": nsr_id
,
2348 "lcmOperationType": operation
,
2350 "isAutomaticInvocation": False,
2351 "operationParams": params
,
2352 "isCancelPending": False,
2354 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2355 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2360 def _format_additional_params(self
, params
):
2361 params
= params
or {}
2362 for key
, value
in params
.items():
2363 if str(value
).startswith("!!yaml "):
2364 params
[key
] = yaml
.safe_load(value
[7:])
2367 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2368 primitive
= seq
.get('name')
2369 primitive_params
= {}
2371 "member_vnf_index": vnf_index
,
2372 "primitive": primitive
,
2373 "primitive_params": primitive_params
,
2376 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2380 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2381 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2382 if op
.get('operationState') == 'COMPLETED':
2383 # b. Skip sub-operation
2384 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2385 return self
.SUBOPERATION_STATUS_SKIP
2387 # c. Reintent executing sub-operation
2388 # The sub-operation exists, and operationState != 'COMPLETED'
2389 # Update operationState = 'PROCESSING' to indicate a reintent.
2390 operationState
= 'PROCESSING'
2391 detailed_status
= 'In progress'
2392 self
._update
_suboperation
_status
(
2393 db_nslcmop
, op_index
, operationState
, detailed_status
)
2394 # Return the sub-operation index
2395 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2396 # with arguments extracted from the sub-operation
2399 # Find a sub-operation where all keys in a matching dictionary must match
2400 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2401 def _find_suboperation(self
, db_nslcmop
, match
):
2402 if (db_nslcmop
and match
):
2403 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2404 for i
, op
in enumerate(op_list
):
2405 if all(op
.get(k
) == match
[k
] for k
in match
):
2407 return self
.SUBOPERATION_STATUS_NOT_FOUND
2409 # Update status for a sub-operation given its index
2410 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2411 # Update DB for HA tasks
2412 q_filter
= {'_id': db_nslcmop
['_id']}
2413 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2414 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2415 self
.db
.set_one("nslcmops",
2417 update_dict
=update_dict
,
2418 fail_on_empty
=False)
2420 # Add sub-operation, return the index of the added sub-operation
2421 # Optionally, set operationState, detailed-status, and operationType
2422 # Status and type are currently set for 'scale' sub-operations:
2423 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2424 # 'detailed-status' : status message
2425 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2426 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2427 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2428 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2429 RO_nsr_id
=None, RO_scaling_info
=None):
2431 return self
.SUBOPERATION_STATUS_NOT_FOUND
2432 # Get the "_admin.operations" list, if it exists
2433 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2434 op_list
= db_nslcmop_admin
.get('operations')
2435 # Create or append to the "_admin.operations" list
2436 new_op
= {'member_vnf_index': vnf_index
,
2438 'vdu_count_index': vdu_count_index
,
2439 'primitive': primitive
,
2440 'primitive_params': mapped_primitive_params
}
2442 new_op
['operationState'] = operationState
2444 new_op
['detailed-status'] = detailed_status
2446 new_op
['lcmOperationType'] = operationType
2448 new_op
['RO_nsr_id'] = RO_nsr_id
2450 new_op
['RO_scaling_info'] = RO_scaling_info
2452 # No existing operations, create key 'operations' with current operation as first list element
2453 db_nslcmop_admin
.update({'operations': [new_op
]})
2454 op_list
= db_nslcmop_admin
.get('operations')
2456 # Existing operations, append operation to list
2457 op_list
.append(new_op
)
2459 db_nslcmop_update
= {'_admin.operations': op_list
}
2460 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2461 op_index
= len(op_list
) - 1
2464 # Helper methods for scale() sub-operations
2466 # pre-scale/post-scale:
2467 # Check for 3 different cases:
2468 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2469 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2470 # c. Reintent: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2471 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2472 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2473 # Find this sub-operation
2474 if (RO_nsr_id
and RO_scaling_info
):
2475 operationType
= 'SCALE-RO'
2477 'member_vnf_index': vnf_index
,
2478 'RO_nsr_id': RO_nsr_id
,
2479 'RO_scaling_info': RO_scaling_info
,
2483 'member_vnf_index': vnf_index
,
2484 'primitive': vnf_config_primitive
,
2485 'primitive_params': primitive_params
,
2486 'lcmOperationType': operationType
2488 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2489 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2490 # a. New sub-operation
2491 # The sub-operation does not exist, add it.
2492 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2493 # The following parameters are set to None for all kind of scaling:
2495 vdu_count_index
= None
2497 if RO_nsr_id
and RO_scaling_info
:
2498 vnf_config_primitive
= None
2499 primitive_params
= None
2502 RO_scaling_info
= None
2503 # Initial status for sub-operation
2504 operationState
= 'PROCESSING'
2505 detailed_status
= 'In progress'
2506 # Add sub-operation for pre/post-scaling (zero or more operations)
2507 self
._add
_suboperation
(db_nslcmop
,
2512 vnf_config_primitive
,
2519 return self
.SUBOPERATION_STATUS_NEW
2521 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2522 # or op_index (operationState != 'COMPLETED')
2523 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2525 # Function to return execution_environment id
2527 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2528 # TODO vdu_index_count
2529 for vca
in vca_deployed_list
:
2530 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2533 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
, vca_index
, destroy_ee
=True):
2535 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2536 :param logging_text:
2538 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2539 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2540 :param vca_index: index in the database _admin.deployed.VCA
2541 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2542 :return: None or exception
2544 # execute terminate_primitives
2545 terminate_primitives
= config_descriptor
.get("terminate-config-primitive")
2546 vdu_id
= vca_deployed
.get("vdu_id")
2547 vdu_count_index
= vca_deployed
.get("vdu_count_index")
2548 vdu_name
= vca_deployed
.get("vdu_name")
2549 vnf_index
= vca_deployed
.get("member-vnf-index")
2550 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
2551 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2552 terminate_primitives
= sorted(terminate_primitives
, key
=lambda x
: int(x
['seq']))
2553 for seq
in terminate_primitives
:
2554 # For each sequence in list, get primitive and call _ns_execute_primitive()
2555 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
2556 vnf_index
, seq
.get("name"))
2557 self
.logger
.debug(logging_text
+ step
)
2558 # Create the primitive for each sequence, i.e. "primitive": "touch"
2559 primitive
= seq
.get('name')
2560 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
2561 # The following 3 parameters are currently set to None for 'terminate':
2562 # vdu_id, vdu_count_index, vdu_name
2565 self
._add
_suboperation
(db_nslcmop
,
2571 mapped_primitive_params
)
2572 # Sub-operations: Call _ns_execute_primitive() instead of action()
2574 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
2575 mapped_primitive_params
)
2576 except LcmException
:
2577 # this happens when VCA is not deployed. In this case it is not needed to terminate
2579 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
2580 if result
not in result_ok
:
2581 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
2582 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
2583 # set that this VCA do not need terminated
2584 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
2585 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
2588 await self
.n2vc
.delete_execution_environment(vca_deployed
["ee_id"])
2590 async def _delete_all_N2VC(self
, db_nsr
: dict):
2591 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
2592 namespace
= "." + db_nsr
["_id"]
2594 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
2595 except N2VCNotFound
: # already deleted. Skip
2597 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
2599 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
2601 Terminates a deployment from RO
2602 :param logging_text:
2603 :param nsr_deployed: db_nsr._admin.deployed
2606 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
2607 this method will update only the index 2, but it will write on database the concatenated content of the list
2612 ro_nsr_id
= ro_delete_action
= None
2613 if nsr_deployed
and nsr_deployed
.get("RO"):
2614 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
2615 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
2618 stage
[2] = "Deleting ns from VIM."
2619 db_nsr_update
["detailed-status"] = " ".join(stage
)
2620 self
._write
_op
_status
(nslcmop_id
, stage
)
2621 self
.logger
.debug(logging_text
+ stage
[2])
2622 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2623 self
._write
_op
_status
(nslcmop_id
, stage
)
2624 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
2625 ro_delete_action
= desc
["action_id"]
2626 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
2627 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2628 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2629 if ro_delete_action
:
2630 # wait until NS is deleted from VIM
2631 stage
[2] = "Waiting ns deleted from VIM."
2632 detailed_status_old
= None
2633 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
2635 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2636 self
._write
_op
_status
(nslcmop_id
, stage
)
2638 delete_timeout
= 20 * 60 # 20 minutes
2639 while delete_timeout
> 0:
2640 desc
= await self
.RO
.show(
2642 item_id_name
=ro_nsr_id
,
2643 extra_item
="action",
2644 extra_item_id
=ro_delete_action
)
2647 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
2649 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
2650 if ns_status
== "ERROR":
2651 raise ROclient
.ROClientException(ns_status_info
)
2652 elif ns_status
== "BUILD":
2653 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
2654 elif ns_status
== "ACTIVE":
2655 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2656 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2659 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
2660 if stage
[2] != detailed_status_old
:
2661 detailed_status_old
= stage
[2]
2662 db_nsr_update
["detailed-status"] = " ".join(stage
)
2663 self
._write
_op
_status
(nslcmop_id
, stage
)
2664 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2665 await asyncio
.sleep(5, loop
=self
.loop
)
2667 else: # delete_timeout <= 0:
2668 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
2670 except Exception as e
:
2671 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2672 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2673 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2674 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2675 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2676 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
2677 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2678 failed_detail
.append("delete conflict: {}".format(e
))
2679 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
2681 failed_detail
.append("delete error: {}".format(e
))
2682 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
2685 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
2686 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
2688 stage
[2] = "Deleting nsd from RO."
2689 db_nsr_update
["detailed-status"] = " ".join(stage
)
2690 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2691 self
._write
_op
_status
(nslcmop_id
, stage
)
2692 await self
.RO
.delete("nsd", ro_nsd_id
)
2693 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
2694 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2695 except Exception as e
:
2696 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2697 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2698 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
2699 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2700 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
2701 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2703 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
2704 self
.logger
.error(logging_text
+ failed_detail
[-1])
2706 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
2707 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
2708 if not vnf_deployed
or not vnf_deployed
["id"]:
2711 ro_vnfd_id
= vnf_deployed
["id"]
2712 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
2713 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
2714 db_nsr_update
["detailed-status"] = " ".join(stage
)
2715 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2716 self
._write
_op
_status
(nslcmop_id
, stage
)
2717 await self
.RO
.delete("vnfd", ro_vnfd_id
)
2718 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
2719 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2720 except Exception as e
:
2721 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2722 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2723 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
2724 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2725 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
2726 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2728 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
2729 self
.logger
.error(logging_text
+ failed_detail
[-1])
2732 stage
[2] = "Error deleting from VIM"
2734 stage
[2] = "Deleted from VIM"
2735 db_nsr_update
["detailed-status"] = " ".join(stage
)
2736 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2737 self
._write
_op
_status
(nslcmop_id
, stage
)
2740 raise LcmException("; ".join(failed_detail
))
2742 async def terminate(self
, nsr_id
, nslcmop_id
):
2743 # Try to lock HA task here
2744 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
2745 if not task_is_locked_by_me
:
2748 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
2749 self
.logger
.debug(logging_text
+ "Enter")
2750 timeout_ns_terminate
= self
.timeout_ns_terminate
2754 error_list
= [] # annotates all failed error messages
2755 db_nslcmop_update
= {}
2756 autoremove
= False # autoremove after terminated
2757 tasks_dict_info
= {}
2759 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
2760 # ^ contains [stage, step, VIM-status]
2762 # wait for any previous tasks in process
2763 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
2765 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2766 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2767 operation_params
= db_nslcmop
.get("operationParams") or {}
2768 if operation_params
.get("timeout_ns_terminate"):
2769 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
2770 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2771 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2773 db_nsr_update
["operational-status"] = "terminating"
2774 db_nsr_update
["config-status"] = "terminating"
2775 self
._write
_ns
_status
(
2777 ns_state
="TERMINATING",
2778 current_operation
="TERMINATING",
2779 current_operation_id
=nslcmop_id
,
2780 other_update
=db_nsr_update
2782 self
._write
_op
_status
(
2787 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
2788 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
2791 stage
[1] = "Getting vnf descriptors from db."
2792 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2793 db_vnfds_from_id
= {}
2794 db_vnfds_from_member_index
= {}
2796 for vnfr
in db_vnfrs_list
:
2797 vnfd_id
= vnfr
["vnfd-id"]
2798 if vnfd_id
not in db_vnfds_from_id
:
2799 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2800 db_vnfds_from_id
[vnfd_id
] = vnfd
2801 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
2803 # Destroy individual execution environments when there are terminating primitives.
2804 # Rest of EE will be deleted at once
2805 if not operation_params
.get("skip_terminate_primitives"):
2806 stage
[0] = "Stage 2/3 execute terminating primitives."
2807 stage
[1] = "Looking execution environment that needs terminate."
2808 self
.logger
.debug(logging_text
+ stage
[1])
2809 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
2810 config_descriptor
= None
2811 if not vca
or not vca
.get("ee_id") or not vca
.get("needed_terminate"):
2813 if not vca
.get("member-vnf-index"):
2815 config_descriptor
= db_nsr
.get("ns-configuration")
2816 elif vca
.get("vdu_id"):
2817 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2818 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
2820 config_descriptor
= vdud
.get("vdu-configuration")
2821 elif vca
.get("kdu_name"):
2822 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2823 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
2825 config_descriptor
= kdud
.get("kdu-configuration")
2827 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
2828 task
= asyncio
.ensure_future(self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
,
2830 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
2832 # wait for pending tasks of terminate primitives
2834 self
.logger
.debug(logging_text
+ 'Waiting for terminate primitive pending tasks...')
2835 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
2836 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
2839 return # raise LcmException("; ".join(error_list))
2840 tasks_dict_info
.clear()
2842 # remove All execution environments at once
2843 stage
[0] = "Stage 3/3 delete all."
2845 if nsr_deployed
.get("VCA"):
2846 stage
[1] = "Deleting all execution environments."
2847 self
.logger
.debug(logging_text
+ stage
[1])
2848 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
2849 timeout
=self
.timeout_charm_delete
))
2850 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
2851 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
2853 # Delete from k8scluster
2854 stage
[1] = "Deleting KDUs."
2855 self
.logger
.debug(logging_text
+ stage
[1])
2856 # print(nsr_deployed)
2857 for kdu
in get_iterable(nsr_deployed
, "K8s"):
2858 if not kdu
or not kdu
.get("kdu-instance"):
2860 kdu_instance
= kdu
.get("kdu-instance")
2861 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
2862 task_delete_kdu_instance
= asyncio
.ensure_future(
2863 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
2864 cluster_uuid
=kdu
.get("k8scluster-uuid"),
2865 kdu_instance
=kdu_instance
))
2867 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
2868 format(kdu
.get("k8scluster-type")))
2870 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
2873 stage
[1] = "Deleting ns from VIM."
2874 task_delete_ro
= asyncio
.ensure_future(
2875 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
2876 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
2878 # rest of staff will be done at finally
2880 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2881 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
2883 except asyncio
.CancelledError
:
2884 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2885 exc
= "Operation was cancelled"
2886 except Exception as e
:
2887 exc
= traceback
.format_exc()
2888 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2891 error_list
.append(str(exc
))
2893 # wait for pending tasks
2895 stage
[1] = "Waiting for terminate pending tasks."
2896 self
.logger
.debug(logging_text
+ stage
[1])
2897 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
2899 stage
[1] = stage
[2] = ""
2900 except asyncio
.CancelledError
:
2901 error_list
.append("Cancelled")
2902 # TODO cancell all tasks
2903 except Exception as exc
:
2904 error_list
.append(str(exc
))
2905 # update status at database
2907 error_detail
= "; ".join(error_list
)
2908 # self.logger.error(logging_text + error_detail)
2909 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
2910 error_description_nsr
= 'Operation: TERMINATING.{}, Stage {}.'.format(nslcmop_id
, stage
[0])
2912 db_nsr_update
["operational-status"] = "failed"
2913 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2914 db_nslcmop_update
["detailed-status"] = error_detail
2915 nslcmop_operation_state
= "FAILED"
2919 error_description_nsr
= error_description_nslcmop
= None
2920 ns_state
= "NOT_INSTANTIATED"
2921 db_nsr_update
["operational-status"] = "terminated"
2922 db_nsr_update
["detailed-status"] = "Done"
2923 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
2924 db_nslcmop_update
["detailed-status"] = "Done"
2925 nslcmop_operation_state
= "COMPLETED"
2928 self
._write
_ns
_status
(
2931 current_operation
="IDLE",
2932 current_operation_id
=None,
2933 error_description
=error_description_nsr
,
2934 error_detail
=error_detail
,
2935 other_update
=db_nsr_update
2938 self
._write
_op
_status
(
2941 error_message
=error_description_nslcmop
,
2942 operation_state
=nslcmop_operation_state
,
2943 other_update
=db_nslcmop_update
,
2945 autoremove
= operation_params
.get("autoremove", False)
2946 if nslcmop_operation_state
:
2948 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2949 "operationState": nslcmop_operation_state
,
2950 "autoremove": autoremove
},
2952 except Exception as e
:
2953 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2955 self
.logger
.debug(logging_text
+ "Exit")
2956 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
2958 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
2960 error_detail_list
= []
2962 pending_tasks
= list(created_tasks_info
.keys())
2963 num_tasks
= len(pending_tasks
)
2965 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
2966 self
._write
_op
_status
(nslcmop_id
, stage
)
2967 while pending_tasks
:
2969 _timeout
= timeout
+ time_start
- time()
2970 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
2971 return_when
=asyncio
.FIRST_COMPLETED
)
2972 num_done
+= len(done
)
2973 if not done
: # Timeout
2974 for task
in pending_tasks
:
2975 new_error
= created_tasks_info
[task
] + ": Timeout"
2976 error_detail_list
.append(new_error
)
2977 error_list
.append(new_error
)
2980 if task
.cancelled():
2983 exc
= task
.exception()
2985 if isinstance(exc
, asyncio
.TimeoutError
):
2987 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
2988 error_list
.append(created_tasks_info
[task
])
2989 error_detail_list
.append(new_error
)
2990 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
2992 self
.logger
.error(logging_text
+ new_error
)
2994 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
2995 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
2997 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
2998 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3000 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
3001 if nsr_id
: # update also nsr
3002 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
3003 "errorDetail": ". ".join(error_detail_list
)})
3004 self
._write
_op
_status
(nslcmop_id
, stage
)
3005 return error_detail_list
3008 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
3010 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
3011 The default-value is used. If it is between < > it look for a value at instantiation_params
3012 :param primitive_desc: portion of VNFD/NSD that describes primitive
3013 :param params: Params provided by user
3014 :param instantiation_params: Instantiation params provided by user
3015 :return: a dictionary with the calculated params
3017 calculated_params
= {}
3018 for parameter
in primitive_desc
.get("parameter", ()):
3019 param_name
= parameter
["name"]
3020 if param_name
in params
:
3021 calculated_params
[param_name
] = params
[param_name
]
3022 elif "default-value" in parameter
or "value" in parameter
:
3023 if "value" in parameter
:
3024 calculated_params
[param_name
] = parameter
["value"]
3026 calculated_params
[param_name
] = parameter
["default-value"]
3027 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3028 and calculated_params
[param_name
].endswith(">"):
3029 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3030 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3032 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3033 format(calculated_params
[param_name
], primitive_desc
["name"]))
3035 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3036 format(param_name
, primitive_desc
["name"]))
3038 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3039 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3041 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3042 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3044 # add always ns_config_info if primitive name is config
3045 if primitive_desc
["name"] == "config":
3046 if "ns_config_info" in instantiation_params
:
3047 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3048 return calculated_params
3050 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None):
3051 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3052 for vca
in deployed_vca
:
3055 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3057 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3059 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3063 # vca_deployed not found
3064 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} is not "
3065 "deployed".format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3068 ee_id
= vca
.get("ee_id")
3070 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3071 "execution environment"
3072 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3075 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3076 retries_interval
=30, timeout
=None) -> (str, str):
3078 if primitive
== "config":
3079 primitive_params
= {"params": primitive_params
}
3083 output
= await asyncio
.wait_for(
3084 self
.n2vc
.exec_primitive(
3086 primitive_name
=primitive
,
3087 params_dict
=primitive_params
,
3088 progress_timeout
=self
.timeout_progress_primitive
,
3089 total_timeout
=self
.timeout_primitive
),
3090 timeout
=timeout
or self
.timeout_primitive
)
3093 except asyncio
.CancelledError
:
3095 except Exception as e
: # asyncio.TimeoutError
3096 if isinstance(e
, asyncio
.TimeoutError
):
3100 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3102 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3104 return 'FAILED', str(e
)
3106 return 'COMPLETED', output
3108 except (LcmException
, asyncio
.CancelledError
):
3110 except Exception as e
:
3111 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3113 async def action(self
, nsr_id
, nslcmop_id
):
3115 # Try to lock HA task here
3116 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3117 if not task_is_locked_by_me
:
3120 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3121 self
.logger
.debug(logging_text
+ "Enter")
3122 # get all needed from database
3126 db_nslcmop_update
= {}
3127 nslcmop_operation_state
= None
3128 error_description_nslcmop
= None
3131 # wait for any previous tasks in process
3132 step
= "Waiting for previous operations to terminate"
3133 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3135 self
._write
_ns
_status
(
3138 current_operation
="RUNNING ACTION",
3139 current_operation_id
=nslcmop_id
3142 step
= "Getting information from database"
3143 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3144 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3146 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3147 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3148 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3149 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3150 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3151 primitive
= db_nslcmop
["operationParams"]["primitive"]
3152 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3153 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3156 step
= "Getting vnfr from database"
3157 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3158 step
= "Getting vnfd from database"
3159 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3161 step
= "Getting nsd from database"
3162 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3164 # for backward compatibility
3165 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3166 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3167 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3168 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3170 # look for primitive
3171 config_primitive_desc
= None
3173 for vdu
in get_iterable(db_vnfd
, "vdu"):
3174 if vdu_id
== vdu
["id"]:
3175 for config_primitive
in deep_get(vdu
, ("vdu-configuration", "config-primitive"), ()):
3176 if config_primitive
["name"] == primitive
:
3177 config_primitive_desc
= config_primitive
3181 for kdu
in get_iterable(db_vnfd
, "kdu"):
3182 if kdu_name
== kdu
["name"]:
3183 for config_primitive
in deep_get(kdu
, ("kdu-configuration", "config-primitive"), ()):
3184 if config_primitive
["name"] == primitive
:
3185 config_primitive_desc
= config_primitive
3189 for config_primitive
in deep_get(db_vnfd
, ("vnf-configuration", "config-primitive"), ()):
3190 if config_primitive
["name"] == primitive
:
3191 config_primitive_desc
= config_primitive
3194 for config_primitive
in deep_get(db_nsd
, ("ns-configuration", "config-primitive"), ()):
3195 if config_primitive
["name"] == primitive
:
3196 config_primitive_desc
= config_primitive
3199 if not config_primitive_desc
and not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3200 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3205 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3206 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3208 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3209 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3211 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3213 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3216 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3218 # TODO check if ns is in a proper status
3219 if kdu_name
and (primitive
in ("upgrade", "rollback", "status") or kdu_action
):
3220 # kdur and desc_params already set from before
3221 if primitive_params
:
3222 desc_params
.update(primitive_params
)
3223 # TODO Check if we will need something at vnf level
3224 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3225 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3228 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3230 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3231 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3232 raise LcmException(msg
)
3234 db_dict
= {"collection": "nsrs",
3235 "filter": {"_id": nsr_id
},
3236 "path": "_admin.deployed.K8s.{}".format(index
)}
3237 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive
, vnf_index
, kdu_name
))
3238 step
= "Executing kdu {}".format(primitive
)
3239 if primitive
== "upgrade":
3240 if desc_params
.get("kdu_model"):
3241 kdu_model
= desc_params
.get("kdu_model")
3242 del desc_params
["kdu_model"]
3244 kdu_model
= kdu
.get("kdu-model")
3245 parts
= kdu_model
.split(sep
=":")
3247 kdu_model
= parts
[0]
3249 detailed_status
= await asyncio
.wait_for(
3250 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3251 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3252 kdu_instance
=kdu
.get("kdu-instance"),
3253 atomic
=True, kdu_model
=kdu_model
,
3254 params
=desc_params
, db_dict
=db_dict
,
3255 timeout
=timeout_ns_action
),
3256 timeout
=timeout_ns_action
+ 10)
3257 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3258 elif primitive
== "rollback":
3259 detailed_status
= await asyncio
.wait_for(
3260 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3261 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3262 kdu_instance
=kdu
.get("kdu-instance"),
3264 timeout
=timeout_ns_action
)
3265 elif primitive
== "status":
3266 detailed_status
= await asyncio
.wait_for(
3267 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3268 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3269 kdu_instance
=kdu
.get("kdu-instance")),
3270 timeout
=timeout_ns_action
)
3272 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3273 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3275 detailed_status
= await asyncio
.wait_for(
3276 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3277 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3278 kdu_instance
=kdu_instance
,
3279 primitive_name
=primitive
,
3280 params
=params
, db_dict
=db_dict
,
3281 timeout
=timeout_ns_action
),
3282 timeout
=timeout_ns_action
)
3285 nslcmop_operation_state
= 'COMPLETED'
3287 detailed_status
= ''
3288 nslcmop_operation_state
= 'FAILED'
3290 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3291 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3292 member_vnf_index
=vnf_index
,
3294 vdu_count_index
=vdu_count_index
),
3295 primitive
=primitive
,
3296 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3297 timeout
=timeout_ns_action
)
3299 db_nslcmop_update
["detailed-status"] = detailed_status
3300 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3301 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3303 return # database update is called inside finally
3305 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3306 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3308 except asyncio
.CancelledError
:
3309 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3310 exc
= "Operation was cancelled"
3311 except asyncio
.TimeoutError
:
3312 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3314 except Exception as e
:
3315 exc
= traceback
.format_exc()
3316 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3319 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3320 "FAILED {}: {}".format(step
, exc
)
3321 nslcmop_operation_state
= "FAILED"
3323 self
._write
_ns
_status
(
3325 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3326 current_operation
="IDLE",
3327 current_operation_id
=None,
3328 # error_description=error_description_nsr,
3329 # error_detail=error_detail,
3330 other_update
=db_nsr_update
3334 self
._write
_op
_status
(
3337 error_message
=error_description_nslcmop
,
3338 operation_state
=nslcmop_operation_state
,
3339 other_update
=db_nslcmop_update
,
3342 if nslcmop_operation_state
:
3344 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3345 "operationState": nslcmop_operation_state
},
3347 except Exception as e
:
3348 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3349 self
.logger
.debug(logging_text
+ "Exit")
3350 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3351 return nslcmop_operation_state
, detailed_status
3353 async def scale(self
, nsr_id
, nslcmop_id
):
3355 # Try to lock HA task here
3356 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3357 if not task_is_locked_by_me
:
3360 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3361 self
.logger
.debug(logging_text
+ "Enter")
3362 # get all needed from database
3365 db_nslcmop_update
= {}
3366 nslcmop_operation_state
= None
3369 # in case of error, indicates what part of scale was failed to put nsr at error status
3370 scale_process
= None
3371 old_operational_status
= ""
3372 old_config_status
= ""
3375 # wait for any previous tasks in process
3376 step
= "Waiting for previous operations to terminate"
3377 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3379 self
._write
_ns
_status
(
3382 current_operation
="SCALING",
3383 current_operation_id
=nslcmop_id
3386 step
= "Getting nslcmop from database"
3387 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3388 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3389 step
= "Getting nsr from database"
3390 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3392 old_operational_status
= db_nsr
["operational-status"]
3393 old_config_status
= db_nsr
["config-status"]
3394 step
= "Parsing scaling parameters"
3395 # self.logger.debug(step)
3396 db_nsr_update
["operational-status"] = "scaling"
3397 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3398 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3401 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3402 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3403 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3404 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3405 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3408 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3409 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3410 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3411 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3412 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3414 # for backward compatibility
3415 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3416 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3417 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3418 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3420 step
= "Getting vnfr from database"
3421 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3422 step
= "Getting vnfd from database"
3423 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3425 step
= "Getting scaling-group-descriptor"
3426 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3427 if scaling_descriptor
["name"] == scaling_group
:
3430 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3431 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3434 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3435 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3436 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3439 # TODO check if ns is in a proper status
3440 step
= "Sending scale order to VIM"
3442 if not db_nsr
["_admin"].get("scaling-group"):
3443 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3444 admin_scale_index
= 0
3446 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3447 if admin_scale_info
["name"] == scaling_group
:
3448 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3450 else: # not found, set index one plus last element and add new entry with the name
3451 admin_scale_index
+= 1
3452 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3453 RO_scaling_info
= []
3454 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3455 if scaling_type
== "SCALE_OUT":
3456 # count if max-instance-count is reached
3457 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3458 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3459 if nb_scale_op
>= max_instance_count
:
3460 raise LcmException("reached the limit of {} (max-instance-count) "
3461 "scaling-out operations for the "
3462 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3465 vdu_scaling_info
["scaling_direction"] = "OUT"
3466 vdu_scaling_info
["vdu-create"] = {}
3467 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3468 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3469 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3470 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3472 elif scaling_type
== "SCALE_IN":
3473 # count if min-instance-count is reached
3474 min_instance_count
= 0
3475 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3476 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3477 if nb_scale_op
<= min_instance_count
:
3478 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3479 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3481 vdu_scaling_info
["scaling_direction"] = "IN"
3482 vdu_scaling_info
["vdu-delete"] = {}
3483 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3484 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3485 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3486 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3488 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3489 vdu_create
= vdu_scaling_info
.get("vdu-create")
3490 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3491 if vdu_scaling_info
["scaling_direction"] == "IN":
3492 for vdur
in reversed(db_vnfr
["vdur"]):
3493 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3494 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3495 vdu_scaling_info
["vdu"].append({
3496 "name": vdur
["name"],
3497 "vdu_id": vdur
["vdu-id-ref"],
3500 for interface
in vdur
["interfaces"]:
3501 vdu_scaling_info
["vdu"][-1]["interface"].append({
3502 "name": interface
["name"],
3503 "ip_address": interface
["ip-address"],
3504 "mac_address": interface
.get("mac-address"),
3506 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
3509 step
= "Executing pre-scale vnf-config-primitive"
3510 if scaling_descriptor
.get("scaling-config-action"):
3511 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3512 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
3513 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
3514 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3515 step
= db_nslcmop_update
["detailed-status"] = \
3516 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3518 # look for primitive
3519 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3520 if config_primitive
["name"] == vnf_config_primitive
:
3524 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
3525 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
3526 "primitive".format(scaling_group
, config_primitive
))
3528 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3529 if db_vnfr
.get("additionalParamsForVnf"):
3530 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3532 scale_process
= "VCA"
3533 db_nsr_update
["config-status"] = "configuring pre-scaling"
3534 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3536 # Pre-scale reintent check: Check if this sub-operation has been executed before
3537 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3538 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
3539 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3540 # Skip sub-operation
3541 result
= 'COMPLETED'
3542 result_detail
= 'Done'
3543 self
.logger
.debug(logging_text
+
3544 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
3545 vnf_config_primitive
, result
, result_detail
))
3547 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3548 # New sub-operation: Get index of this sub-operation
3549 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3550 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3551 format(vnf_config_primitive
))
3553 # Reintent: Get registered params for this existing sub-operation
3554 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3555 vnf_index
= op
.get('member_vnf_index')
3556 vnf_config_primitive
= op
.get('primitive')
3557 primitive_params
= op
.get('primitive_params')
3558 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3559 format(vnf_config_primitive
))
3560 # Execute the primitive, either with new (first-time) or registered (reintent) args
3561 result
, result_detail
= await self
._ns
_execute
_primitive
(
3562 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3563 member_vnf_index
=vnf_index
,
3565 vdu_count_index
=None),
3566 vnf_config_primitive
, primitive_params
)
3567 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3568 vnf_config_primitive
, result
, result_detail
))
3569 # Update operationState = COMPLETED | FAILED
3570 self
._update
_suboperation
_status
(
3571 db_nslcmop
, op_index
, result
, result_detail
)
3573 if result
== "FAILED":
3574 raise LcmException(result_detail
)
3575 db_nsr_update
["config-status"] = old_config_status
3576 scale_process
= None
3580 # Should this block be skipped if 'RO_nsr_id' == None ?
3581 # if (RO_nsr_id and RO_scaling_info):
3583 scale_process
= "RO"
3584 # Scale RO reintent check: Check if this sub-operation has been executed before
3585 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3586 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
3587 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3588 # Skip sub-operation
3589 result
= 'COMPLETED'
3590 result_detail
= 'Done'
3591 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
3592 result
, result_detail
))
3594 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3595 # New sub-operation: Get index of this sub-operation
3596 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3597 self
.logger
.debug(logging_text
+ "New sub-operation RO")
3599 # Reintent: Get registered params for this existing sub-operation
3600 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3601 RO_nsr_id
= op
.get('RO_nsr_id')
3602 RO_scaling_info
= op
.get('RO_scaling_info')
3603 self
.logger
.debug(logging_text
+ "Sub-operation RO reintent".format(
3604 vnf_config_primitive
))
3606 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
3607 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
3608 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
3610 RO_nslcmop_id
= RO_desc
["instance_action_id"]
3611 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
3613 RO_task_done
= False
3614 step
= detailed_status
= "Waiting RO_task_id={} to complete the scale action.".format(RO_nslcmop_id
)
3615 detailed_status_old
= None
3616 self
.logger
.debug(logging_text
+ step
)
3618 deployment_timeout
= 1 * 3600 # One hour
3619 while deployment_timeout
> 0:
3620 if not RO_task_done
:
3621 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
3622 extra_item_id
=RO_nslcmop_id
)
3625 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3627 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3628 if ns_status
== "ERROR":
3629 raise ROclient
.ROClientException(ns_status_info
)
3630 elif ns_status
== "BUILD":
3631 detailed_status
= step
+ "; {}".format(ns_status_info
)
3632 elif ns_status
== "ACTIVE":
3634 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
3635 self
.logger
.debug(logging_text
+ step
)
3637 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3640 if ns_status
== "ERROR":
3641 raise ROclient
.ROClientException(ns_status_info
)
3642 elif ns_status
== "BUILD":
3643 detailed_status
= step
+ "; {}".format(ns_status_info
)
3644 elif ns_status
== "ACTIVE":
3645 step
= detailed_status
= \
3646 "Waiting for management IP address reported by the VIM. Updating VNFRs"
3648 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
3651 desc
= await self
.RO
.show("ns", RO_nsr_id
)
3654 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3656 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
3657 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
3659 except LcmExceptionNoMgmtIP
:
3662 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
3663 if detailed_status
!= detailed_status_old
:
3664 self
._update
_suboperation
_status
(
3665 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
3666 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
3667 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3669 await asyncio
.sleep(5, loop
=self
.loop
)
3670 deployment_timeout
-= 5
3671 if deployment_timeout
<= 0:
3672 self
._update
_suboperation
_status
(
3673 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
3674 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
3676 # update VDU_SCALING_INFO with the obtained ip_addresses
3677 if vdu_scaling_info
["scaling_direction"] == "OUT":
3678 for vdur
in reversed(db_vnfr
["vdur"]):
3679 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
3680 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
3681 vdu_scaling_info
["vdu"].append({
3682 "name": vdur
["name"],
3683 "vdu_id": vdur
["vdu-id-ref"],
3686 for interface
in vdur
["interfaces"]:
3687 vdu_scaling_info
["vdu"][-1]["interface"].append({
3688 "name": interface
["name"],
3689 "ip_address": interface
["ip-address"],
3690 "mac_address": interface
.get("mac-address"),
3692 del vdu_scaling_info
["vdu-create"]
3694 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
3697 scale_process
= None
3699 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3702 # execute primitive service POST-SCALING
3703 step
= "Executing post-scale vnf-config-primitive"
3704 if scaling_descriptor
.get("scaling-config-action"):
3705 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3706 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
3707 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
3708 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3709 step
= db_nslcmop_update
["detailed-status"] = \
3710 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3712 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3713 if db_vnfr
.get("additionalParamsForVnf"):
3714 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3716 # look for primitive
3717 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3718 if config_primitive
["name"] == vnf_config_primitive
:
3721 raise LcmException("Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:"
3722 "scaling-config-action[vnf-config-primitive-name-ref='{}'] does not "
3723 "match any vnf-configuration:config-primitive".format(scaling_group
,
3725 scale_process
= "VCA"
3726 db_nsr_update
["config-status"] = "configuring post-scaling"
3727 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3729 # Post-scale reintent check: Check if this sub-operation has been executed before
3730 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3731 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
3732 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3733 # Skip sub-operation
3734 result
= 'COMPLETED'
3735 result_detail
= 'Done'
3736 self
.logger
.debug(logging_text
+
3737 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
3738 format(vnf_config_primitive
, result
, result_detail
))
3740 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3741 # New sub-operation: Get index of this sub-operation
3742 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3743 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3744 format(vnf_config_primitive
))
3746 # Reintent: Get registered params for this existing sub-operation
3747 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3748 vnf_index
= op
.get('member_vnf_index')
3749 vnf_config_primitive
= op
.get('primitive')
3750 primitive_params
= op
.get('primitive_params')
3751 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3752 format(vnf_config_primitive
))
3753 # Execute the primitive, either with new (first-time) or registered (reintent) args
3754 result
, result_detail
= await self
._ns
_execute
_primitive
(
3755 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3756 member_vnf_index
=vnf_index
,
3758 vdu_count_index
=None),
3759 vnf_config_primitive
, primitive_params
)
3760 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3761 vnf_config_primitive
, result
, result_detail
))
3762 # Update operationState = COMPLETED | FAILED
3763 self
._update
_suboperation
_status
(
3764 db_nslcmop
, op_index
, result
, result_detail
)
3766 if result
== "FAILED":
3767 raise LcmException(result_detail
)
3768 db_nsr_update
["config-status"] = old_config_status
3769 scale_process
= None
3772 db_nslcmop_update
["operationState"] = nslcmop_operation_state
= "COMPLETED"
3773 db_nslcmop_update
["statusEnteredTime"] = time()
3774 db_nslcmop_update
["detailed-status"] = "done"
3775 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
3776 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
3777 else old_operational_status
3778 db_nsr_update
["config-status"] = old_config_status
3780 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
3781 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3783 except asyncio
.CancelledError
:
3784 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3785 exc
= "Operation was cancelled"
3786 except Exception as e
:
3787 exc
= traceback
.format_exc()
3788 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3790 self
._write
_ns
_status
(
3793 current_operation
="IDLE",
3794 current_operation_id
=None
3798 db_nslcmop_update
["detailed-status"] = "FAILED {}: {}".format(step
, exc
)
3799 db_nslcmop_update
["operationState"] = nslcmop_operation_state
= "FAILED"
3800 db_nslcmop_update
["statusEnteredTime"] = time()
3802 db_nsr_update
["operational-status"] = old_operational_status
3803 db_nsr_update
["config-status"] = old_config_status
3804 db_nsr_update
["detailed-status"] = ""
3806 if "VCA" in scale_process
:
3807 db_nsr_update
["config-status"] = "failed"
3808 if "RO" in scale_process
:
3809 db_nsr_update
["operational-status"] = "failed"
3810 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
3813 if db_nslcmop
and db_nslcmop_update
:
3814 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3816 self
._write
_ns
_status
(
3819 current_operation
="IDLE",
3820 current_operation_id
=None,
3821 other_update
=db_nsr_update
3824 except DbException
as e
:
3825 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
3826 if nslcmop_operation_state
:
3828 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3829 "operationState": nslcmop_operation_state
},
3832 # await asyncio.sleep(cooldown_time, loop=self.loop)
3833 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
3834 except Exception as e
:
3835 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3836 self
.logger
.debug(logging_text
+ "Exit")
3837 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")