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 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
789 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
790 for vnfr
in db_vnfrs
.values():
791 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
794 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
797 # get vnfds, instantiate at RO
798 for c_vnf
in nsd
.get("constituent-vnfd", ()):
799 member_vnf_index
= c_vnf
["member-vnf-index"]
800 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
801 vnfd_ref
= vnfd
["id"]
803 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
804 db_nsr_update
["detailed-status"] = " ".join(stage
)
805 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
806 self
._write
_op
_status
(nslcmop_id
, stage
)
808 # self.logger.debug(logging_text + stage[2])
809 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
810 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
811 RO_descriptor_number
+= 1
813 # look position at deployed.RO.vnfd if not present it will be appended at the end
814 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
815 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
818 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
819 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
822 RO_update
= {"member-vnf-index": member_vnf_index
}
823 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
825 RO_update
["id"] = vnfd_list
[0]["uuid"]
826 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
827 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
829 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
830 get("additionalParamsForVnf"), nsr_id
)
831 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
832 RO_update
["id"] = desc
["uuid"]
833 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
834 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
835 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
836 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
841 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
842 db_nsr_update
["detailed-status"] = " ".join(stage
)
843 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
844 self
._write
_op
_status
(nslcmop_id
, stage
)
846 # self.logger.debug(logging_text + stage[2])
847 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
848 RO_descriptor_number
+= 1
849 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
851 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
852 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
853 nsd_ref
, RO_nsd_uuid
))
855 nsd_RO
= deepcopy(nsd
)
856 nsd_RO
["id"] = RO_osm_nsd_id
857 nsd_RO
.pop("_id", None)
858 nsd_RO
.pop("_admin", None)
859 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
860 member_vnf_index
= c_vnf
["member-vnf-index"]
861 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
862 for c_vld
in nsd_RO
.get("vld", ()):
863 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
864 member_vnf_index
= cp
["member-vnf-index-ref"]
865 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
867 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
868 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
869 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
870 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
871 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
874 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
875 db_nsr_update
["detailed-status"] = " ".join(stage
)
876 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
877 self
._write
_op
_status
(nslcmop_id
, stage
)
879 # if present use it unless in error status
880 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
883 stage
[2] = "Looking for existing ns at RO"
884 db_nsr_update
["detailed-status"] = " ".join(stage
)
885 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
886 self
._write
_op
_status
(nslcmop_id
, stage
)
887 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
888 desc
= await self
.RO
.show("ns", RO_nsr_id
)
890 except ROclient
.ROClientException
as e
:
891 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
893 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
895 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
896 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
897 if ns_status
== "ERROR":
898 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
899 self
.logger
.debug(logging_text
+ stage
[2])
900 await self
.RO
.delete("ns", RO_nsr_id
)
901 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
903 stage
[2] = "Checking dependencies"
904 db_nsr_update
["detailed-status"] = " ".join(stage
)
905 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
906 self
._write
_op
_status
(nslcmop_id
, stage
)
907 # self.logger.debug(logging_text + stage[2])
909 # check if VIM is creating and wait look if previous tasks in process
910 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
912 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
913 self
.logger
.debug(logging_text
+ stage
[2])
914 await asyncio
.wait(task_dependency
, timeout
=3600)
915 if ns_params
.get("vnf"):
916 for vnf
in ns_params
["vnf"]:
917 if "vimAccountId" in vnf
:
918 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
921 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
922 self
.logger
.debug(logging_text
+ stage
[2])
923 await asyncio
.wait(task_dependency
, timeout
=3600)
925 stage
[2] = "Checking instantiation parameters."
926 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, db_vnfrs
, n2vc_key_list
)
927 stage
[2] = "Deploying ns at VIM."
928 db_nsr_update
["detailed-status"] = " ".join(stage
)
929 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
930 self
._write
_op
_status
(nslcmop_id
, stage
)
932 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
933 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
934 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
935 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
936 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
938 # wait until NS is ready
939 stage
[2] = "Waiting VIM to deploy ns."
940 db_nsr_update
["detailed-status"] = " ".join(stage
)
941 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
942 self
._write
_op
_status
(nslcmop_id
, stage
)
943 detailed_status_old
= None
944 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
947 while time() <= start_deploy
+ timeout_ns_deploy
:
948 desc
= await self
.RO
.show("ns", RO_nsr_id
)
952 # desc has changed => update db
953 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
956 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
957 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
958 if ns_status
== "ERROR":
959 raise ROclient
.ROClientException(ns_status_info
)
960 elif ns_status
== "BUILD":
961 stage
[2] = "VIM: ({})".format(ns_status_info
)
962 elif ns_status
== "ACTIVE":
963 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
965 self
.ns_update_vnfr(db_vnfrs
, desc
)
967 except LcmExceptionNoMgmtIP
:
970 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
971 if stage
[2] != detailed_status_old
:
972 detailed_status_old
= stage
[2]
973 db_nsr_update
["detailed-status"] = " ".join(stage
)
974 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
975 self
._write
_op
_status
(nslcmop_id
, stage
)
976 await asyncio
.sleep(5, loop
=self
.loop
)
977 else: # timeout_ns_deploy
978 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
981 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
983 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
984 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
985 stage
[2] = "Deployed at VIM"
986 db_nsr_update
["detailed-status"] = " ".join(stage
)
987 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
988 self
._write
_op
_status
(nslcmop_id
, stage
)
989 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
990 # self.logger.debug(logging_text + "Deployed at VIM")
991 except (ROclient
.ROClientException
, LcmException
, DbException
) as e
:
992 stage
[2] = "ERROR deploying at VIM"
993 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
996 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
998 Wait for ip addres at RO, and optionally, insert public key in virtual machine
999 :param logging_text: prefix use for logging
1004 :param pub_key: public ssh key to inject, None to skip
1005 :param user: user to apply the public ssh key
1009 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
1013 target_vdu_id
= None
1019 if ro_retries
>= 360: # 1 hour
1020 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
1022 await asyncio
.sleep(10, loop
=self
.loop
)
1025 if not target_vdu_id
:
1026 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1028 if not vdu_id
: # for the VNF case
1029 if db_vnfr
.get("status") == "ERROR":
1030 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1031 ip_address
= db_vnfr
.get("ip-address")
1034 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1036 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1037 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1039 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1040 vdur
= db_vnfr
["vdur"][0]
1042 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1045 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1046 ip_address
= vdur
.get("ip-address")
1049 target_vdu_id
= vdur
["vdu-id-ref"]
1050 elif vdur
.get("status") == "ERROR":
1051 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1053 if not target_vdu_id
:
1056 # inject public key into machine
1057 if pub_key
and user
:
1058 # wait until NS is deployed at RO
1060 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1061 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1065 # self.logger.debug(logging_text + "Inserting RO key")
1066 if vdur
.get("pdu-type"):
1067 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1070 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1071 result_dict
= await self
.RO
.create_action(
1073 item_id_name
=ro_nsr_id
,
1074 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1076 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1077 if not result_dict
or not isinstance(result_dict
, dict):
1078 raise LcmException("Unknown response from RO when injecting key")
1079 for result
in result_dict
.values():
1080 if result
.get("vim_result") == 200:
1083 raise ROclient
.ROClientException("error injecting key: {}".format(
1084 result
.get("description")))
1086 except ROclient
.ROClientException
as e
:
1088 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1092 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1098 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1100 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1102 my_vca
= vca_deployed_list
[vca_index
]
1103 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1104 # vdu or kdu: no dependencies
1108 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1109 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1110 configuration_status_list
= db_nsr
["configurationStatus"]
1111 for index
, vca_deployed
in enumerate(configuration_status_list
):
1112 if index
== vca_index
:
1115 if not my_vca
.get("member-vnf-index") or \
1116 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1117 internal_status
= configuration_status_list
[index
].get("status")
1118 if internal_status
== 'READY':
1120 elif internal_status
== 'BROKEN':
1121 raise LcmException("Configuration aborted because dependent charm/s has failed")
1125 # no dependencies, return
1127 await asyncio
.sleep(10)
1130 raise LcmException("Configuration aborted because dependent charm/s timeout")
1132 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1133 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
):
1134 nsr_id
= db_nsr
["_id"]
1135 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1136 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1137 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1139 'collection': 'nsrs',
1140 'filter': {'_id': nsr_id
},
1141 'path': db_update_entry
1147 element_under_configuration
= nsr_id
1151 vnfr_id
= db_vnfr
["_id"]
1153 namespace
= "{nsi}.{ns}".format(
1154 nsi
=nsi_id
if nsi_id
else "",
1158 element_type
= 'VNF'
1159 element_under_configuration
= vnfr_id
1160 namespace
+= ".{}".format(vnfr_id
)
1162 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1163 element_type
= 'VDU'
1164 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1166 namespace
+= ".{}".format(kdu_name
)
1167 element_type
= 'KDU'
1168 element_under_configuration
= kdu_name
1171 self
.fs
.sync() # Sync from FSMongo
1172 artifact_path
= "{}/{}/charms/{}".format(
1173 base_folder
["folder"],
1174 base_folder
["pkg-dir"],
1175 config_descriptor
["juju"]["charm"]
1178 is_proxy_charm
= deep_get(config_descriptor
, ('juju', 'charm')) is not None
1179 if deep_get(config_descriptor
, ('juju', 'proxy')) is False:
1180 is_proxy_charm
= False
1182 # n2vc_redesign STEP 3.1
1184 # find old ee_id if exists
1185 ee_id
= vca_deployed
.get("ee_id")
1187 # create or register execution environment in VCA
1190 self
._write
_configuration
_status
(
1192 vca_index
=vca_index
,
1194 element_under_configuration
=element_under_configuration
,
1195 element_type
=element_type
1198 step
= "create execution environment"
1199 self
.logger
.debug(logging_text
+ step
)
1200 ee_id
, credentials
= await self
.n2vc
.create_execution_environment(namespace
=namespace
,
1205 step
= "Waiting to VM being up and getting IP address"
1206 self
.logger
.debug(logging_text
+ step
)
1207 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1208 user
=None, pub_key
=None)
1209 credentials
= {"hostname": rw_mgmt_ip
}
1211 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1212 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1213 # merged. Meanwhile let's get username from initial-config-primitive
1214 if not username
and config_descriptor
.get("initial-config-primitive"):
1215 for config_primitive
in config_descriptor
["initial-config-primitive"]:
1216 for param
in config_primitive
.get("parameter", ()):
1217 if param
["name"] == "ssh-username":
1218 username
= param
["value"]
1221 raise LcmException("Cannot determine the username neither with 'initial-config-promitive' nor with "
1222 "'config-access.ssh-access.default-user'")
1223 credentials
["username"] = username
1224 # n2vc_redesign STEP 3.2
1226 self
._write
_configuration
_status
(
1228 vca_index
=vca_index
,
1229 status
='REGISTERING',
1230 element_under_configuration
=element_under_configuration
,
1231 element_type
=element_type
1234 step
= "register execution environment {}".format(credentials
)
1235 self
.logger
.debug(logging_text
+ step
)
1236 ee_id
= await self
.n2vc
.register_execution_environment(credentials
=credentials
, namespace
=namespace
,
1239 # for compatibility with MON/POL modules, the need model and application name at database
1240 # TODO ask to N2VC instead of assuming the format "model_name.application_name"
1241 ee_id_parts
= ee_id
.split('.')
1242 model_name
= ee_id_parts
[0]
1243 application_name
= ee_id_parts
[1]
1244 db_nsr_update
= {db_update_entry
+ "model": model_name
,
1245 db_update_entry
+ "application": application_name
,
1246 db_update_entry
+ "ee_id": ee_id
}
1248 # n2vc_redesign STEP 3.3
1250 step
= "Install configuration Software"
1252 self
._write
_configuration
_status
(
1254 vca_index
=vca_index
,
1255 status
='INSTALLING SW',
1256 element_under_configuration
=element_under_configuration
,
1257 element_type
=element_type
,
1258 other_update
=db_nsr_update
1261 # TODO check if already done
1262 self
.logger
.debug(logging_text
+ step
)
1264 if not is_proxy_charm
:
1265 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1266 if initial_config_primitive_list
:
1267 for primitive
in initial_config_primitive_list
:
1268 if primitive
["name"] == "config":
1269 config
= self
._map
_primitive
_params
(
1275 await self
.n2vc
.install_configuration_sw(
1277 artifact_path
=artifact_path
,
1282 # write in db flag of configuration_sw already installed
1283 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1285 # add relations for this VCA (wait for other peers related with this VCA)
1286 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
, vca_index
=vca_index
)
1288 # if SSH access is required, then get execution environment SSH public
1289 if is_proxy_charm
: # if native charm we have waited already to VM be UP
1292 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1293 # Needed to inject a ssh key
1294 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1295 step
= "Install configuration Software, getting public ssh key"
1296 pub_key
= await self
.n2vc
.get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1298 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1300 step
= "Waiting to VM being up and getting IP address"
1301 self
.logger
.debug(logging_text
+ step
)
1303 # n2vc_redesign STEP 5.1
1304 # wait for RO (ip-address) Insert pub_key into VM
1306 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1307 user
=user
, pub_key
=pub_key
)
1309 rw_mgmt_ip
= None # This is for a NS configuration
1311 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1313 # store rw_mgmt_ip in deploy params for later replacement
1314 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1316 # n2vc_redesign STEP 6 Execute initial config primitive
1317 step
= 'execute initial config primitive'
1318 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1320 # sort initial config primitives by 'seq'
1321 if initial_config_primitive_list
:
1323 initial_config_primitive_list
.sort(key
=lambda val
: int(val
['seq']))
1324 except Exception as e
:
1325 self
.logger
.error(logging_text
+ step
+ ": " + str(e
))
1327 self
.logger
.debug(logging_text
+ step
+ ": No initial-config-primitive")
1329 # add config if not present for NS charm
1330 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1333 # wait for dependent primitives execution (NS -> VNF -> VDU)
1334 if initial_config_primitive_list
:
1335 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1337 # stage, in function of element type: vdu, kdu, vnf or ns
1338 my_vca
= vca_deployed_list
[vca_index
]
1339 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1341 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1342 elif my_vca
.get("member-vnf-index"):
1344 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1347 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1349 self
._write
_configuration
_status
(
1351 vca_index
=vca_index
,
1352 status
='EXECUTING PRIMITIVE'
1355 self
._write
_op
_status
(
1360 check_if_terminated_needed
= True
1361 for initial_config_primitive
in initial_config_primitive_list
:
1362 # adding information on the vca_deployed if it is a NS execution environment
1363 if not vca_deployed
["member-vnf-index"]:
1364 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1365 # TODO check if already done
1366 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1368 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1369 self
.logger
.debug(logging_text
+ step
)
1370 await self
.n2vc
.exec_primitive(
1372 primitive_name
=initial_config_primitive
["name"],
1373 params_dict
=primitive_params_
,
1376 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1377 if check_if_terminated_needed
:
1378 if config_descriptor
.get('terminate-config-primitive'):
1379 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1380 check_if_terminated_needed
= False
1382 # TODO register in database that primitive is done
1384 step
= "instantiated at VCA"
1385 self
.logger
.debug(logging_text
+ step
)
1387 self
._write
_configuration
_status
(
1389 vca_index
=vca_index
,
1393 except Exception as e
: # TODO not use Exception but N2VC exception
1394 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1395 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1396 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1397 self
._write
_configuration
_status
(
1399 vca_index
=vca_index
,
1402 raise LcmException("{} {}".format(step
, e
)) from e
1404 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1405 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1407 Update db_nsr fields.
1410 :param current_operation:
1411 :param current_operation_id:
1412 :param error_description:
1413 :param error_detail:
1414 :param other_update: Other required changes at database if provided, will be cleared
1418 db_dict
= other_update
or {}
1419 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1420 db_dict
["_admin.current-operation"] = current_operation_id
1421 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1422 db_dict
["currentOperation"] = current_operation
1423 db_dict
["currentOperationID"] = current_operation_id
1424 db_dict
["errorDescription"] = error_description
1425 db_dict
["errorDetail"] = error_detail
1428 db_dict
["nsState"] = ns_state
1429 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1430 except DbException
as e
:
1431 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1433 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1434 operation_state
: str = None, other_update
: dict = None):
1436 db_dict
= other_update
or {}
1437 db_dict
['queuePosition'] = queuePosition
1438 if isinstance(stage
, list):
1439 db_dict
['stage'] = stage
[0]
1440 db_dict
['detailed-status'] = " ".join(stage
)
1441 elif stage
is not None:
1442 db_dict
['stage'] = str(stage
)
1444 if error_message
is not None:
1445 db_dict
['errorMessage'] = error_message
1446 if operation_state
is not None:
1447 db_dict
['operationState'] = operation_state
1448 db_dict
["statusEnteredTime"] = time()
1449 self
.update_db_2("nslcmops", op_id
, db_dict
)
1450 except DbException
as e
:
1451 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1453 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1455 nsr_id
= db_nsr
["_id"]
1456 # configurationStatus
1457 config_status
= db_nsr
.get('configurationStatus')
1459 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1460 enumerate(config_status
) if v
}
1462 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1464 except DbException
as e
:
1465 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1467 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1468 element_under_configuration
: str = None, element_type
: str = None,
1469 other_update
: dict = None):
1471 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1472 # .format(vca_index, status))
1475 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1476 db_dict
= other_update
or {}
1478 db_dict
[db_path
+ 'status'] = status
1479 if element_under_configuration
:
1480 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1482 db_dict
[db_path
+ 'elementType'] = element_type
1483 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1484 except DbException
as e
:
1485 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1486 .format(status
, nsr_id
, vca_index
, e
))
1488 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1490 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1491 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1492 Database is used because the result can be obtained from a different LCM worker in case of HA.
1493 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1494 :param db_nslcmop: database content of nslcmop
1495 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1496 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
1497 computed 'vim-account-id'
1500 nslcmop_id
= db_nslcmop
['_id']
1501 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1502 if placement_engine
== "PLA":
1503 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1504 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1505 db_poll_interval
= 5
1506 wait
= db_poll_interval
* 10
1508 while not pla_result
and wait
>= 0:
1509 await asyncio
.sleep(db_poll_interval
)
1510 wait
-= db_poll_interval
1511 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1512 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1515 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1517 for pla_vnf
in pla_result
['vnf']:
1518 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1519 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1522 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1524 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1527 def update_nsrs_with_pla_result(self
, params
):
1529 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1530 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1531 except Exception as e
:
1532 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1534 async def instantiate(self
, nsr_id
, nslcmop_id
):
1537 :param nsr_id: ns instance to deploy
1538 :param nslcmop_id: operation to run
1542 # Try to lock HA task here
1543 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1544 if not task_is_locked_by_me
:
1545 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1548 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1549 self
.logger
.debug(logging_text
+ "Enter")
1551 # get all needed from database
1553 # database nsrs record
1556 # database nslcmops record
1559 # update operation on nsrs
1561 # update operation on nslcmops
1562 db_nslcmop_update
= {}
1564 nslcmop_operation_state
= None
1565 db_vnfrs
= {} # vnf's info indexed by member-index
1567 tasks_dict_info
= {} # from task to info text
1570 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1571 # ^ stage, step, VIM progress
1573 # wait for any previous tasks in process
1574 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1576 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1577 stage
[1] = "Reading from database,"
1578 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1579 db_nsr_update
["detailed-status"] = "creating"
1580 db_nsr_update
["operational-status"] = "init"
1581 self
._write
_ns
_status
(
1583 ns_state
="BUILDING",
1584 current_operation
="INSTANTIATING",
1585 current_operation_id
=nslcmop_id
,
1586 other_update
=db_nsr_update
1588 self
._write
_op
_status
(
1594 # read from db: operation
1595 stage
[1] = "Getting nslcmop={} from db".format(nslcmop_id
)
1596 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1597 ns_params
= db_nslcmop
.get("operationParams")
1598 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1599 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1601 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1604 stage
[1] = "Getting nsr={} from db".format(nsr_id
)
1605 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1606 # nsd is replicated into ns (no db read)
1608 # nsr_name = db_nsr["name"] # TODO short-name??
1610 # read from db: vnf's of this ns
1611 stage
[1] = "Getting vnfrs from db"
1612 self
.logger
.debug(logging_text
+ stage
[1])
1613 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1615 # read from db: vnfd's for every vnf
1616 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1617 db_vnfds
= {} # every vnfd data indexed by vnf id
1618 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1620 # for each vnf in ns, read vnfd
1621 for vnfr
in db_vnfrs_list
:
1622 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1623 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1624 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1625 # if we haven't this vnfd, read it from db
1626 if vnfd_id
not in db_vnfds
:
1628 stage
[1] = "Getting vnfd={} id='{}' from db".format(vnfd_id
, vnfd_ref
)
1629 self
.logger
.debug(logging_text
+ stage
[1])
1630 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1633 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1634 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1635 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1637 # Get or generates the _admin.deployed.VCA list
1638 vca_deployed_list
= None
1639 if db_nsr
["_admin"].get("deployed"):
1640 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1641 if vca_deployed_list
is None:
1642 vca_deployed_list
= []
1643 configuration_status_list
= []
1644 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1645 db_nsr_update
["configurationStatus"] = configuration_status_list
1646 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1647 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1648 elif isinstance(vca_deployed_list
, dict):
1649 # maintain backward compatibility. Change a dict to list at database
1650 vca_deployed_list
= list(vca_deployed_list
.values())
1651 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1652 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1654 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1655 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1656 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1658 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1659 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1660 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1662 # n2vc_redesign STEP 2 Deploy Network Scenario
1663 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1664 self
._write
_op
_status
(
1669 stage
[1] = "Deploying KDUs,"
1670 # self.logger.debug(logging_text + "Before deploy_kdus")
1671 # Call to deploy_kdus in case exists the "vdu:kdu" param
1672 await self
.deploy_kdus(
1673 logging_text
=logging_text
,
1675 nslcmop_id
=nslcmop_id
,
1678 task_instantiation_info
=tasks_dict_info
,
1681 stage
[1] = "Getting VCA public key."
1682 # n2vc_redesign STEP 1 Get VCA public ssh-key
1683 # feature 1429. Add n2vc public key to needed VMs
1684 n2vc_key
= self
.n2vc
.get_public_key()
1685 n2vc_key_list
= [n2vc_key
]
1686 if self
.vca_config
.get("public_key"):
1687 n2vc_key_list
.append(self
.vca_config
["public_key"])
1689 stage
[1] = "Deploying NS at VIM."
1690 task_ro
= asyncio
.ensure_future(
1691 self
.instantiate_RO(
1692 logging_text
=logging_text
,
1696 db_nslcmop
=db_nslcmop
,
1698 db_vnfds_ref
=db_vnfds_ref
,
1699 n2vc_key_list
=n2vc_key_list
,
1703 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
1704 tasks_dict_info
[task_ro
] = "Deploying at VIM"
1706 # n2vc_redesign STEP 3 to 6 Deploy N2VC
1707 stage
[1] = "Deploying Execution Environments."
1708 self
.logger
.debug(logging_text
+ stage
[1])
1710 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
1711 # get_iterable() returns a value from a dict or empty tuple if key does not exist
1712 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
1713 vnfd_id
= c_vnf
["vnfd-id-ref"]
1714 vnfd
= db_vnfds_ref
[vnfd_id
]
1715 member_vnf_index
= str(c_vnf
["member-vnf-index"])
1716 db_vnfr
= db_vnfrs
[member_vnf_index
]
1717 base_folder
= vnfd
["_admin"]["storage"]
1723 # Get additional parameters
1725 if db_vnfr
.get("additionalParamsForVnf"):
1726 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
1728 descriptor_config
= vnfd
.get("vnf-configuration")
1729 if descriptor_config
and descriptor_config
.get("juju"):
1731 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
1734 nslcmop_id
=nslcmop_id
,
1740 member_vnf_index
=member_vnf_index
,
1741 vdu_index
=vdu_index
,
1743 deploy_params
=deploy_params
,
1744 descriptor_config
=descriptor_config
,
1745 base_folder
=base_folder
,
1746 task_instantiation_info
=tasks_dict_info
,
1750 # Deploy charms for each VDU that supports one.
1751 for vdud
in get_iterable(vnfd
, 'vdu'):
1753 descriptor_config
= vdud
.get('vdu-configuration')
1754 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
1755 if vdur
.get("additionalParams"):
1756 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
1758 deploy_params_vdu
= deploy_params
1759 if descriptor_config
and descriptor_config
.get("juju"):
1760 # look for vdu index in the db_vnfr["vdu"] section
1761 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1762 # if vdur["vdu-id-ref"] == vdu_id:
1765 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1766 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1767 # vdu_name = vdur.get("name")
1770 for vdu_index
in range(int(vdud
.get("count", 1))):
1771 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
1773 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
1774 member_vnf_index
, vdu_id
, vdu_index
),
1777 nslcmop_id
=nslcmop_id
,
1783 member_vnf_index
=member_vnf_index
,
1784 vdu_index
=vdu_index
,
1786 deploy_params
=deploy_params_vdu
,
1787 descriptor_config
=descriptor_config
,
1788 base_folder
=base_folder
,
1789 task_instantiation_info
=tasks_dict_info
,
1792 for kdud
in get_iterable(vnfd
, 'kdu'):
1793 kdu_name
= kdud
["name"]
1794 descriptor_config
= kdud
.get('kdu-configuration')
1795 if descriptor_config
and descriptor_config
.get("juju"):
1799 # look for vdu index in the db_vnfr["vdu"] section
1800 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1801 # if vdur["vdu-id-ref"] == vdu_id:
1804 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1805 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1806 # vdu_name = vdur.get("name")
1810 logging_text
=logging_text
,
1813 nslcmop_id
=nslcmop_id
,
1819 member_vnf_index
=member_vnf_index
,
1820 vdu_index
=vdu_index
,
1822 deploy_params
=deploy_params
,
1823 descriptor_config
=descriptor_config
,
1824 base_folder
=base_folder
,
1825 task_instantiation_info
=tasks_dict_info
,
1829 # Check if this NS has a charm configuration
1830 descriptor_config
= nsd
.get("ns-configuration")
1831 if descriptor_config
and descriptor_config
.get("juju"):
1834 member_vnf_index
= None
1840 # Get additional parameters
1842 if db_nsr
.get("additionalParamsForNs"):
1843 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
1844 base_folder
= nsd
["_admin"]["storage"]
1846 logging_text
=logging_text
,
1849 nslcmop_id
=nslcmop_id
,
1855 member_vnf_index
=member_vnf_index
,
1856 vdu_index
=vdu_index
,
1858 deploy_params
=deploy_params
,
1859 descriptor_config
=descriptor_config
,
1860 base_folder
=base_folder
,
1861 task_instantiation_info
=tasks_dict_info
,
1865 # rest of staff will be done at finally
1867 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
1868 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
1870 except asyncio
.CancelledError
:
1871 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
1872 exc
= "Operation was cancelled"
1873 except Exception as e
:
1874 exc
= traceback
.format_exc()
1875 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
1878 error_list
.append(str(exc
))
1880 # wait for pending tasks
1882 stage
[1] = "Waiting for instantiate pending tasks."
1883 self
.logger
.debug(logging_text
+ stage
[1])
1884 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
1885 stage
, nslcmop_id
, nsr_id
=nsr_id
)
1886 stage
[1] = stage
[2] = ""
1887 except asyncio
.CancelledError
:
1888 error_list
.append("Cancelled")
1889 # TODO cancel all tasks
1890 except Exception as exc
:
1891 error_list
.append(str(exc
))
1893 # update operation-status
1894 db_nsr_update
["operational-status"] = "running"
1895 # let's begin with VCA 'configured' status (later we can change it)
1896 db_nsr_update
["config-status"] = "configured"
1897 for task
, task_name
in tasks_dict_info
.items():
1898 if not task
.done() or task
.cancelled() or task
.exception():
1899 if task_name
.startswith(self
.task_name_deploy_vca
):
1900 # A N2VC task is pending
1901 db_nsr_update
["config-status"] = "failed"
1903 # RO or KDU task is pending
1904 db_nsr_update
["operational-status"] = "failed"
1906 # update status at database
1908 error_detail
= ". ".join(error_list
)
1909 self
.logger
.error(logging_text
+ error_detail
)
1910 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
1911 error_description_nsr
= 'Operation: INSTANTIATING.{}, Stage {}'.format(nslcmop_id
, stage
[0])
1913 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
1914 db_nslcmop_update
["detailed-status"] = error_detail
1915 nslcmop_operation_state
= "FAILED"
1919 error_description_nsr
= error_description_nslcmop
= None
1921 db_nsr_update
["detailed-status"] = "Done"
1922 db_nslcmop_update
["detailed-status"] = "Done"
1923 nslcmop_operation_state
= "COMPLETED"
1926 self
._write
_ns
_status
(
1929 current_operation
="IDLE",
1930 current_operation_id
=None,
1931 error_description
=error_description_nsr
,
1932 error_detail
=error_detail
,
1933 other_update
=db_nsr_update
1935 self
._write
_op
_status
(
1938 error_message
=error_description_nslcmop
,
1939 operation_state
=nslcmop_operation_state
,
1940 other_update
=db_nslcmop_update
,
1943 if nslcmop_operation_state
:
1945 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
1946 "operationState": nslcmop_operation_state
},
1948 except Exception as e
:
1949 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
1951 self
.logger
.debug(logging_text
+ "Exit")
1952 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
1954 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int, timeout
: int = 3600) -> bool:
1957 # 1. find all relations for this VCA
1958 # 2. wait for other peers related
1963 # STEP 1: find all relations for this VCA
1966 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1969 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
1971 # read all ns-configuration relations
1972 ns_relations
= list()
1973 db_ns_relations
= deep_get(db_nsr
, ('nsd', 'ns-configuration', 'relation'))
1975 for r
in db_ns_relations
:
1976 # check if this VCA is in the relation
1977 if my_vca
.get('member-vnf-index') in\
1978 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1979 ns_relations
.append(r
)
1981 # read all vnf-configuration relations
1982 vnf_relations
= list()
1983 db_vnfd_list
= db_nsr
.get('vnfd-id')
1985 for vnfd
in db_vnfd_list
:
1986 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
1987 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
1988 if db_vnf_relations
:
1989 for r
in db_vnf_relations
:
1990 # check if this VCA is in the relation
1991 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
1992 vnf_relations
.append(r
)
1994 # if no relations, terminate
1995 if not ns_relations
and not vnf_relations
:
1996 self
.logger
.debug(logging_text
+ ' No relations')
1999 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
2006 if now
- start
>= timeout
:
2007 self
.logger
.error(logging_text
+ ' : timeout adding relations')
2010 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2011 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2013 # for each defined NS relation, find the VCA's related
2014 for r
in ns_relations
:
2015 from_vca_ee_id
= None
2017 from_vca_endpoint
= None
2018 to_vca_endpoint
= None
2019 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2020 for vca
in vca_list
:
2021 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
2022 and vca
.get('config_sw_installed'):
2023 from_vca_ee_id
= vca
.get('ee_id')
2024 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2025 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2026 and vca
.get('config_sw_installed'):
2027 to_vca_ee_id
= vca
.get('ee_id')
2028 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2029 if from_vca_ee_id
and to_vca_ee_id
:
2031 await self
.n2vc
.add_relation(
2032 ee_id_1
=from_vca_ee_id
,
2033 ee_id_2
=to_vca_ee_id
,
2034 endpoint_1
=from_vca_endpoint
,
2035 endpoint_2
=to_vca_endpoint
)
2036 # remove entry from relations list
2037 ns_relations
.remove(r
)
2039 # check failed peers
2041 vca_status_list
= db_nsr
.get('configurationStatus')
2043 for i
in range(len(vca_list
)):
2045 vca_status
= vca_status_list
[i
]
2046 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2047 if vca_status
.get('status') == 'BROKEN':
2048 # peer broken: remove relation from list
2049 ns_relations
.remove(r
)
2050 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2051 if vca_status
.get('status') == 'BROKEN':
2052 # peer broken: remove relation from list
2053 ns_relations
.remove(r
)
2058 # for each defined VNF relation, find the VCA's related
2059 for r
in vnf_relations
:
2060 from_vca_ee_id
= None
2062 from_vca_endpoint
= None
2063 to_vca_endpoint
= None
2064 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2065 for vca
in vca_list
:
2066 if vca
.get('vdu_id') == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2067 from_vca_ee_id
= vca
.get('ee_id')
2068 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2069 if vca
.get('vdu_id') == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2070 to_vca_ee_id
= vca
.get('ee_id')
2071 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2072 if from_vca_ee_id
and to_vca_ee_id
:
2074 await self
.n2vc
.add_relation(
2075 ee_id_1
=from_vca_ee_id
,
2076 ee_id_2
=to_vca_ee_id
,
2077 endpoint_1
=from_vca_endpoint
,
2078 endpoint_2
=to_vca_endpoint
)
2079 # remove entry from relations list
2080 vnf_relations
.remove(r
)
2082 # check failed peers
2084 vca_status_list
= db_nsr
.get('configurationStatus')
2086 for i
in range(len(vca_list
)):
2088 vca_status
= vca_status_list
[i
]
2089 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2090 if vca_status
.get('status') == 'BROKEN':
2091 # peer broken: remove relation from list
2092 ns_relations
.remove(r
)
2093 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2094 if vca_status
.get('status') == 'BROKEN':
2095 # peer broken: remove relation from list
2096 ns_relations
.remove(r
)
2102 await asyncio
.sleep(5.0)
2104 if not ns_relations
and not vnf_relations
:
2105 self
.logger
.debug('Relations added')
2110 except Exception as e
:
2111 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2114 def _write_db_callback(self
, task
, item
, _id
, on_done
=None, on_exc
=None):
2116 callback for kdu install intended to store the returned kdu_instance at database
2121 result
= task
.result()
2123 db_update
[on_done
] = str(result
)
2124 except Exception as e
:
2126 db_update
[on_exc
] = str(e
)
2129 self
.update_db_2(item
, _id
, db_update
)
2133 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2134 # Launch kdus if present in the descriptor
2136 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2138 def _get_cluster_id(cluster_id
, cluster_type
):
2139 nonlocal k8scluster_id_2_uuic
2140 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2141 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2143 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2144 if not db_k8scluster
:
2145 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2146 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2148 raise LcmException("K8s cluster '{}' has not been initilized for '{}'".format(cluster_id
, cluster_type
))
2149 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2152 logging_text
+= "Deploy kdus: "
2155 db_nsr_update
= {"_admin.deployed.K8s": []}
2156 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2159 updated_cluster_list
= []
2161 for vnfr_data
in db_vnfrs
.values():
2162 for kdur
in get_iterable(vnfr_data
, "kdur"):
2163 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2164 vnfd_id
= vnfr_data
.get('vnfd-id')
2165 namespace
= kdur
.get("k8s-namespace")
2166 if kdur
.get("helm-chart"):
2167 kdumodel
= kdur
["helm-chart"]
2168 k8sclustertype
= "helm-chart"
2169 elif kdur
.get("juju-bundle"):
2170 kdumodel
= kdur
["juju-bundle"]
2171 k8sclustertype
= "juju-bundle"
2173 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2174 "juju-bundle. Maybe an old NBI version is running".
2175 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2176 # check if kdumodel is a file and exists
2178 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2179 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2180 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2181 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["'pkg-dir"], k8sclustertype
,
2183 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2184 kdumodel
= self
.fs
.path
+ filename
2185 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2187 except Exception: # it is not a file
2190 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2191 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2192 cluster_uuid
= _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2194 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2195 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2196 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2197 if del_repo_list
or added_repo_dict
:
2198 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2199 updated
= {'_admin.helm_charts_added.' +
2200 item
: name
for item
, name
in added_repo_dict
.items()}
2201 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2202 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2204 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2205 updated_cluster_list
.append(cluster_uuid
)
2207 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2208 kdur
["kdu-name"], k8s_cluster_id
)
2210 k8s_instace_info
= {"kdu-instance": None,
2211 "k8scluster-uuid": cluster_uuid
,
2212 "k8scluster-type": k8sclustertype
,
2213 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2214 "kdu-name": kdur
["kdu-name"],
2215 "kdu-model": kdumodel
,
2216 "namespace": namespace
}
2217 db_path
= "_admin.deployed.K8s.{}".format(index
)
2218 db_nsr_update
[db_path
] = k8s_instace_info
2219 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2221 db_dict
= {"collection": "nsrs",
2222 "filter": {"_id": nsr_id
},
2225 task
= asyncio
.ensure_future(
2226 self
.k8scluster_map
[k8sclustertype
].install(cluster_uuid
=cluster_uuid
, kdu_model
=kdumodel
,
2227 atomic
=True, params
=desc_params
,
2228 db_dict
=db_dict
, timeout
=600,
2229 kdu_name
=kdur
["kdu-name"], namespace
=namespace
))
2231 task
.add_done_callback(partial(self
._write
_db
_callback
, item
="nsrs", _id
=nsr_id
,
2232 on_done
=db_path
+ ".kdu-instance",
2233 on_exc
=db_path
+ ".detailed-status"))
2234 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2235 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2239 except (LcmException
, asyncio
.CancelledError
):
2241 except Exception as e
:
2242 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2243 if isinstance(e
, (N2VCException
, DbException
)):
2244 self
.logger
.error(logging_text
+ msg
)
2246 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2247 raise LcmException(msg
)
2250 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2252 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2253 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2254 base_folder
, task_instantiation_info
, stage
):
2255 # launch instantiate_N2VC in a asyncio task and register task object
2256 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2257 # if not found, create one entry and update database
2259 # fill db_nsr._admin.deployed.VCA.<index>
2261 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2262 if not vca_deployed
:
2264 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2265 vca_deployed
.get("vdu_id") == vdu_id
and \
2266 vca_deployed
.get("kdu_name") == kdu_name
and \
2267 vca_deployed
.get("vdu_count_index", 0) == vdu_index
:
2270 # not found, create one.
2272 "member-vnf-index": member_vnf_index
,
2274 "kdu_name": kdu_name
,
2275 "vdu_count_index": vdu_index
,
2276 "operational-status": "init", # TODO revise
2277 "detailed-status": "", # TODO revise
2278 "step": "initial-deploy", # TODO revise
2280 "vdu_name": vdu_name
,
2284 # create VCA and configurationStatus in db
2286 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2287 "configurationStatus.{}".format(vca_index
): dict()
2289 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2291 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2294 task_n2vc
= asyncio
.ensure_future(
2295 self
.instantiate_N2VC(
2296 logging_text
=logging_text
,
2297 vca_index
=vca_index
,
2303 vdu_index
=vdu_index
,
2304 deploy_params
=deploy_params
,
2305 config_descriptor
=descriptor_config
,
2306 base_folder
=base_folder
,
2307 nslcmop_id
=nslcmop_id
,
2311 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2312 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2313 member_vnf_index
or "", vdu_id
or "")
2315 # Check if this VNFD has a configured terminate action
2316 def _has_terminate_config_primitive(self
, vnfd
):
2317 vnf_config
= vnfd
.get("vnf-configuration")
2318 if vnf_config
and vnf_config
.get("terminate-config-primitive"):
2324 def _get_terminate_config_primitive_seq_list(vnfd
):
2325 """ Get a numerically sorted list of the sequences for this VNFD's terminate action """
2326 # No need to check for existing primitive twice, already done before
2327 vnf_config
= vnfd
.get("vnf-configuration")
2328 seq_list
= vnf_config
.get("terminate-config-primitive")
2329 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2330 seq_list_sorted
= sorted(seq_list
, key
=lambda x
: int(x
['seq']))
2331 return seq_list_sorted
2334 def _create_nslcmop(nsr_id
, operation
, params
):
2336 Creates a ns-lcm-opp content to be stored at database.
2337 :param nsr_id: internal id of the instance
2338 :param operation: instantiate, terminate, scale, action, ...
2339 :param params: user parameters for the operation
2340 :return: dictionary following SOL005 format
2342 # Raise exception if invalid arguments
2343 if not (nsr_id
and operation
and params
):
2345 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2351 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2352 "operationState": "PROCESSING",
2353 "statusEnteredTime": now
,
2354 "nsInstanceId": nsr_id
,
2355 "lcmOperationType": operation
,
2357 "isAutomaticInvocation": False,
2358 "operationParams": params
,
2359 "isCancelPending": False,
2361 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2362 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2367 def _format_additional_params(self
, params
):
2368 params
= params
or {}
2369 for key
, value
in params
.items():
2370 if str(value
).startswith("!!yaml "):
2371 params
[key
] = yaml
.safe_load(value
[7:])
2374 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2375 primitive
= seq
.get('name')
2376 primitive_params
= {}
2378 "member_vnf_index": vnf_index
,
2379 "primitive": primitive
,
2380 "primitive_params": primitive_params
,
2383 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2387 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2388 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2389 if op
.get('operationState') == 'COMPLETED':
2390 # b. Skip sub-operation
2391 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2392 return self
.SUBOPERATION_STATUS_SKIP
2394 # c. Reintent executing sub-operation
2395 # The sub-operation exists, and operationState != 'COMPLETED'
2396 # Update operationState = 'PROCESSING' to indicate a reintent.
2397 operationState
= 'PROCESSING'
2398 detailed_status
= 'In progress'
2399 self
._update
_suboperation
_status
(
2400 db_nslcmop
, op_index
, operationState
, detailed_status
)
2401 # Return the sub-operation index
2402 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2403 # with arguments extracted from the sub-operation
2406 # Find a sub-operation where all keys in a matching dictionary must match
2407 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2408 def _find_suboperation(self
, db_nslcmop
, match
):
2409 if (db_nslcmop
and match
):
2410 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2411 for i
, op
in enumerate(op_list
):
2412 if all(op
.get(k
) == match
[k
] for k
in match
):
2414 return self
.SUBOPERATION_STATUS_NOT_FOUND
2416 # Update status for a sub-operation given its index
2417 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2418 # Update DB for HA tasks
2419 q_filter
= {'_id': db_nslcmop
['_id']}
2420 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2421 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2422 self
.db
.set_one("nslcmops",
2424 update_dict
=update_dict
,
2425 fail_on_empty
=False)
2427 # Add sub-operation, return the index of the added sub-operation
2428 # Optionally, set operationState, detailed-status, and operationType
2429 # Status and type are currently set for 'scale' sub-operations:
2430 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2431 # 'detailed-status' : status message
2432 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2433 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2434 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2435 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2436 RO_nsr_id
=None, RO_scaling_info
=None):
2438 return self
.SUBOPERATION_STATUS_NOT_FOUND
2439 # Get the "_admin.operations" list, if it exists
2440 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2441 op_list
= db_nslcmop_admin
.get('operations')
2442 # Create or append to the "_admin.operations" list
2443 new_op
= {'member_vnf_index': vnf_index
,
2445 'vdu_count_index': vdu_count_index
,
2446 'primitive': primitive
,
2447 'primitive_params': mapped_primitive_params
}
2449 new_op
['operationState'] = operationState
2451 new_op
['detailed-status'] = detailed_status
2453 new_op
['lcmOperationType'] = operationType
2455 new_op
['RO_nsr_id'] = RO_nsr_id
2457 new_op
['RO_scaling_info'] = RO_scaling_info
2459 # No existing operations, create key 'operations' with current operation as first list element
2460 db_nslcmop_admin
.update({'operations': [new_op
]})
2461 op_list
= db_nslcmop_admin
.get('operations')
2463 # Existing operations, append operation to list
2464 op_list
.append(new_op
)
2466 db_nslcmop_update
= {'_admin.operations': op_list
}
2467 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2468 op_index
= len(op_list
) - 1
2471 # Helper methods for scale() sub-operations
2473 # pre-scale/post-scale:
2474 # Check for 3 different cases:
2475 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2476 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2477 # c. Reintent: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2478 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2479 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2480 # Find this sub-operation
2481 if (RO_nsr_id
and RO_scaling_info
):
2482 operationType
= 'SCALE-RO'
2484 'member_vnf_index': vnf_index
,
2485 'RO_nsr_id': RO_nsr_id
,
2486 'RO_scaling_info': RO_scaling_info
,
2490 'member_vnf_index': vnf_index
,
2491 'primitive': vnf_config_primitive
,
2492 'primitive_params': primitive_params
,
2493 'lcmOperationType': operationType
2495 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2496 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2497 # a. New sub-operation
2498 # The sub-operation does not exist, add it.
2499 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2500 # The following parameters are set to None for all kind of scaling:
2502 vdu_count_index
= None
2504 if RO_nsr_id
and RO_scaling_info
:
2505 vnf_config_primitive
= None
2506 primitive_params
= None
2509 RO_scaling_info
= None
2510 # Initial status for sub-operation
2511 operationState
= 'PROCESSING'
2512 detailed_status
= 'In progress'
2513 # Add sub-operation for pre/post-scaling (zero or more operations)
2514 self
._add
_suboperation
(db_nslcmop
,
2519 vnf_config_primitive
,
2526 return self
.SUBOPERATION_STATUS_NEW
2528 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2529 # or op_index (operationState != 'COMPLETED')
2530 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2532 # Function to return execution_environment id
2534 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2535 # TODO vdu_index_count
2536 for vca
in vca_deployed_list
:
2537 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2540 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
, vca_index
, destroy_ee
=True):
2542 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2543 :param logging_text:
2545 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2546 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2547 :param vca_index: index in the database _admin.deployed.VCA
2548 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2549 :return: None or exception
2551 # execute terminate_primitives
2552 terminate_primitives
= config_descriptor
.get("terminate-config-primitive")
2553 vdu_id
= vca_deployed
.get("vdu_id")
2554 vdu_count_index
= vca_deployed
.get("vdu_count_index")
2555 vdu_name
= vca_deployed
.get("vdu_name")
2556 vnf_index
= vca_deployed
.get("member-vnf-index")
2557 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
2558 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2559 terminate_primitives
= sorted(terminate_primitives
, key
=lambda x
: int(x
['seq']))
2560 for seq
in terminate_primitives
:
2561 # For each sequence in list, get primitive and call _ns_execute_primitive()
2562 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
2563 vnf_index
, seq
.get("name"))
2564 self
.logger
.debug(logging_text
+ step
)
2565 # Create the primitive for each sequence, i.e. "primitive": "touch"
2566 primitive
= seq
.get('name')
2567 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
2568 # The following 3 parameters are currently set to None for 'terminate':
2569 # vdu_id, vdu_count_index, vdu_name
2572 self
._add
_suboperation
(db_nslcmop
,
2578 mapped_primitive_params
)
2579 # Sub-operations: Call _ns_execute_primitive() instead of action()
2581 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
2582 mapped_primitive_params
)
2583 except LcmException
:
2584 # this happens when VCA is not deployed. In this case it is not needed to terminate
2586 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
2587 if result
not in result_ok
:
2588 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
2589 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
2590 # set that this VCA do not need terminated
2591 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
2592 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
2595 await self
.n2vc
.delete_execution_environment(vca_deployed
["ee_id"])
2597 async def _delete_all_N2VC(self
, db_nsr
: dict):
2598 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
2599 namespace
= "." + db_nsr
["_id"]
2601 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
2602 except N2VCNotFound
: # already deleted. Skip
2604 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
2606 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
2608 Terminates a deployment from RO
2609 :param logging_text:
2610 :param nsr_deployed: db_nsr._admin.deployed
2613 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
2614 this method will update only the index 2, but it will write on database the concatenated content of the list
2619 ro_nsr_id
= ro_delete_action
= None
2620 if nsr_deployed
and nsr_deployed
.get("RO"):
2621 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
2622 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
2625 stage
[2] = "Deleting ns from VIM."
2626 db_nsr_update
["detailed-status"] = " ".join(stage
)
2627 self
._write
_op
_status
(nslcmop_id
, stage
)
2628 self
.logger
.debug(logging_text
+ stage
[2])
2629 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2630 self
._write
_op
_status
(nslcmop_id
, stage
)
2631 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
2632 ro_delete_action
= desc
["action_id"]
2633 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
2634 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2635 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2636 if ro_delete_action
:
2637 # wait until NS is deleted from VIM
2638 stage
[2] = "Waiting ns deleted from VIM."
2639 detailed_status_old
= None
2640 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
2642 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2643 self
._write
_op
_status
(nslcmop_id
, stage
)
2645 delete_timeout
= 20 * 60 # 20 minutes
2646 while delete_timeout
> 0:
2647 desc
= await self
.RO
.show(
2649 item_id_name
=ro_nsr_id
,
2650 extra_item
="action",
2651 extra_item_id
=ro_delete_action
)
2654 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
2656 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
2657 if ns_status
== "ERROR":
2658 raise ROclient
.ROClientException(ns_status_info
)
2659 elif ns_status
== "BUILD":
2660 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
2661 elif ns_status
== "ACTIVE":
2662 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2663 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2666 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
2667 if stage
[2] != detailed_status_old
:
2668 detailed_status_old
= stage
[2]
2669 db_nsr_update
["detailed-status"] = " ".join(stage
)
2670 self
._write
_op
_status
(nslcmop_id
, stage
)
2671 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2672 await asyncio
.sleep(5, loop
=self
.loop
)
2674 else: # delete_timeout <= 0:
2675 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
2677 except Exception as e
:
2678 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2679 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2680 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2681 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2682 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2683 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
2684 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2685 failed_detail
.append("delete conflict: {}".format(e
))
2686 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
2688 failed_detail
.append("delete error: {}".format(e
))
2689 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
2692 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
2693 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
2695 stage
[2] = "Deleting nsd from RO."
2696 db_nsr_update
["detailed-status"] = " ".join(stage
)
2697 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2698 self
._write
_op
_status
(nslcmop_id
, stage
)
2699 await self
.RO
.delete("nsd", ro_nsd_id
)
2700 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
2701 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2702 except Exception as e
:
2703 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2704 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2705 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
2706 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2707 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
2708 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2710 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
2711 self
.logger
.error(logging_text
+ failed_detail
[-1])
2713 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
2714 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
2715 if not vnf_deployed
or not vnf_deployed
["id"]:
2718 ro_vnfd_id
= vnf_deployed
["id"]
2719 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
2720 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
2721 db_nsr_update
["detailed-status"] = " ".join(stage
)
2722 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2723 self
._write
_op
_status
(nslcmop_id
, stage
)
2724 await self
.RO
.delete("vnfd", ro_vnfd_id
)
2725 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
2726 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2727 except Exception as e
:
2728 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2729 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2730 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
2731 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2732 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
2733 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2735 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
2736 self
.logger
.error(logging_text
+ failed_detail
[-1])
2739 stage
[2] = "Error deleting from VIM"
2741 stage
[2] = "Deleted from VIM"
2742 db_nsr_update
["detailed-status"] = " ".join(stage
)
2743 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2744 self
._write
_op
_status
(nslcmop_id
, stage
)
2747 raise LcmException("; ".join(failed_detail
))
2749 async def terminate(self
, nsr_id
, nslcmop_id
):
2750 # Try to lock HA task here
2751 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
2752 if not task_is_locked_by_me
:
2755 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
2756 self
.logger
.debug(logging_text
+ "Enter")
2757 timeout_ns_terminate
= self
.timeout_ns_terminate
2760 operation_params
= None
2762 error_list
= [] # annotates all failed error messages
2763 db_nslcmop_update
= {}
2764 autoremove
= False # autoremove after terminated
2765 tasks_dict_info
= {}
2767 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
2768 # ^ contains [stage, step, VIM-status]
2770 # wait for any previous tasks in process
2771 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
2773 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2774 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2775 operation_params
= db_nslcmop
.get("operationParams") or {}
2776 if operation_params
.get("timeout_ns_terminate"):
2777 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
2778 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2779 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2781 db_nsr_update
["operational-status"] = "terminating"
2782 db_nsr_update
["config-status"] = "terminating"
2783 self
._write
_ns
_status
(
2785 ns_state
="TERMINATING",
2786 current_operation
="TERMINATING",
2787 current_operation_id
=nslcmop_id
,
2788 other_update
=db_nsr_update
2790 self
._write
_op
_status
(
2795 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
2796 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
2799 stage
[1] = "Getting vnf descriptors from db."
2800 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2801 db_vnfds_from_id
= {}
2802 db_vnfds_from_member_index
= {}
2804 for vnfr
in db_vnfrs_list
:
2805 vnfd_id
= vnfr
["vnfd-id"]
2806 if vnfd_id
not in db_vnfds_from_id
:
2807 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
2808 db_vnfds_from_id
[vnfd_id
] = vnfd
2809 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
2811 # Destroy individual execution environments when there are terminating primitives.
2812 # Rest of EE will be deleted at once
2813 if not operation_params
.get("skip_terminate_primitives"):
2814 stage
[0] = "Stage 2/3 execute terminating primitives."
2815 stage
[1] = "Looking execution environment that needs terminate."
2816 self
.logger
.debug(logging_text
+ stage
[1])
2817 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
2818 config_descriptor
= None
2819 if not vca
or not vca
.get("ee_id") or not vca
.get("needed_terminate"):
2821 if not vca
.get("member-vnf-index"):
2823 config_descriptor
= db_nsr
.get("ns-configuration")
2824 elif vca
.get("vdu_id"):
2825 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2826 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
2828 config_descriptor
= vdud
.get("vdu-configuration")
2829 elif vca
.get("kdu_name"):
2830 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
2831 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
2833 config_descriptor
= kdud
.get("kdu-configuration")
2835 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
2836 task
= asyncio
.ensure_future(self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
,
2838 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
2840 # wait for pending tasks of terminate primitives
2842 self
.logger
.debug(logging_text
+ 'Waiting for terminate primitive pending tasks...')
2843 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
2844 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
2847 return # raise LcmException("; ".join(error_list))
2848 tasks_dict_info
.clear()
2850 # remove All execution environments at once
2851 stage
[0] = "Stage 3/3 delete all."
2853 if nsr_deployed
.get("VCA"):
2854 stage
[1] = "Deleting all execution environments."
2855 self
.logger
.debug(logging_text
+ stage
[1])
2856 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
2857 timeout
=self
.timeout_charm_delete
))
2858 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
2859 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
2861 # Delete from k8scluster
2862 stage
[1] = "Deleting KDUs."
2863 self
.logger
.debug(logging_text
+ stage
[1])
2864 # print(nsr_deployed)
2865 for kdu
in get_iterable(nsr_deployed
, "K8s"):
2866 if not kdu
or not kdu
.get("kdu-instance"):
2868 kdu_instance
= kdu
.get("kdu-instance")
2869 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
2870 task_delete_kdu_instance
= asyncio
.ensure_future(
2871 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
2872 cluster_uuid
=kdu
.get("k8scluster-uuid"),
2873 kdu_instance
=kdu_instance
))
2875 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
2876 format(kdu
.get("k8scluster-type")))
2878 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
2881 stage
[1] = "Deleting ns from VIM."
2882 task_delete_ro
= asyncio
.ensure_future(
2883 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
2884 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
2886 # rest of staff will be done at finally
2888 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2889 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
2891 except asyncio
.CancelledError
:
2892 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2893 exc
= "Operation was cancelled"
2894 except Exception as e
:
2895 exc
= traceback
.format_exc()
2896 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2899 error_list
.append(str(exc
))
2901 # wait for pending tasks
2903 stage
[1] = "Waiting for terminate pending tasks."
2904 self
.logger
.debug(logging_text
+ stage
[1])
2905 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
2907 stage
[1] = stage
[2] = ""
2908 except asyncio
.CancelledError
:
2909 error_list
.append("Cancelled")
2910 # TODO cancell all tasks
2911 except Exception as exc
:
2912 error_list
.append(str(exc
))
2913 # update status at database
2915 error_detail
= "; ".join(error_list
)
2916 # self.logger.error(logging_text + error_detail)
2917 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
2918 error_description_nsr
= 'Operation: TERMINATING.{}, Stage {}.'.format(nslcmop_id
, stage
[0])
2920 db_nsr_update
["operational-status"] = "failed"
2921 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2922 db_nslcmop_update
["detailed-status"] = error_detail
2923 nslcmop_operation_state
= "FAILED"
2927 error_description_nsr
= error_description_nslcmop
= None
2928 ns_state
= "NOT_INSTANTIATED"
2929 db_nsr_update
["operational-status"] = "terminated"
2930 db_nsr_update
["detailed-status"] = "Done"
2931 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
2932 db_nslcmop_update
["detailed-status"] = "Done"
2933 nslcmop_operation_state
= "COMPLETED"
2936 self
._write
_ns
_status
(
2939 current_operation
="IDLE",
2940 current_operation_id
=None,
2941 error_description
=error_description_nsr
,
2942 error_detail
=error_detail
,
2943 other_update
=db_nsr_update
2945 self
._write
_op
_status
(
2948 error_message
=error_description_nslcmop
,
2949 operation_state
=nslcmop_operation_state
,
2950 other_update
=db_nslcmop_update
,
2952 if operation_params
:
2953 autoremove
= operation_params
.get("autoremove", False)
2954 if nslcmop_operation_state
:
2956 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2957 "operationState": nslcmop_operation_state
,
2958 "autoremove": autoremove
},
2960 except Exception as e
:
2961 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2963 self
.logger
.debug(logging_text
+ "Exit")
2964 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
2966 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
2968 error_detail_list
= []
2970 pending_tasks
= list(created_tasks_info
.keys())
2971 num_tasks
= len(pending_tasks
)
2973 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
2974 self
._write
_op
_status
(nslcmop_id
, stage
)
2975 while pending_tasks
:
2977 _timeout
= timeout
+ time_start
- time()
2978 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
2979 return_when
=asyncio
.FIRST_COMPLETED
)
2980 num_done
+= len(done
)
2981 if not done
: # Timeout
2982 for task
in pending_tasks
:
2983 new_error
= created_tasks_info
[task
] + ": Timeout"
2984 error_detail_list
.append(new_error
)
2985 error_list
.append(new_error
)
2988 if task
.cancelled():
2991 exc
= task
.exception()
2993 if isinstance(exc
, asyncio
.TimeoutError
):
2995 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
2996 error_list
.append(created_tasks_info
[task
])
2997 error_detail_list
.append(new_error
)
2998 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
3000 self
.logger
.error(logging_text
+ new_error
)
3002 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
3003 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
3005 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
3006 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3008 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
3009 if nsr_id
: # update also nsr
3010 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
3011 "errorDetail": ". ".join(error_detail_list
)})
3012 self
._write
_op
_status
(nslcmop_id
, stage
)
3013 return error_detail_list
3016 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
3018 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
3019 The default-value is used. If it is between < > it look for a value at instantiation_params
3020 :param primitive_desc: portion of VNFD/NSD that describes primitive
3021 :param params: Params provided by user
3022 :param instantiation_params: Instantiation params provided by user
3023 :return: a dictionary with the calculated params
3025 calculated_params
= {}
3026 for parameter
in primitive_desc
.get("parameter", ()):
3027 param_name
= parameter
["name"]
3028 if param_name
in params
:
3029 calculated_params
[param_name
] = params
[param_name
]
3030 elif "default-value" in parameter
or "value" in parameter
:
3031 if "value" in parameter
:
3032 calculated_params
[param_name
] = parameter
["value"]
3034 calculated_params
[param_name
] = parameter
["default-value"]
3035 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3036 and calculated_params
[param_name
].endswith(">"):
3037 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3038 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3040 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3041 format(calculated_params
[param_name
], primitive_desc
["name"]))
3043 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3044 format(param_name
, primitive_desc
["name"]))
3046 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3047 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3049 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3050 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3052 # add always ns_config_info if primitive name is config
3053 if primitive_desc
["name"] == "config":
3054 if "ns_config_info" in instantiation_params
:
3055 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3056 return calculated_params
3058 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None):
3059 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3060 for vca
in deployed_vca
:
3063 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3065 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3067 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3071 # vca_deployed not found
3072 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} is not "
3073 "deployed".format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3076 ee_id
= vca
.get("ee_id")
3078 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3079 "execution environment"
3080 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3083 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3084 retries_interval
=30, timeout
=None) -> (str, str):
3086 if primitive
== "config":
3087 primitive_params
= {"params": primitive_params
}
3091 output
= await asyncio
.wait_for(
3092 self
.n2vc
.exec_primitive(
3094 primitive_name
=primitive
,
3095 params_dict
=primitive_params
,
3096 progress_timeout
=self
.timeout_progress_primitive
,
3097 total_timeout
=self
.timeout_primitive
),
3098 timeout
=timeout
or self
.timeout_primitive
)
3101 except asyncio
.CancelledError
:
3103 except Exception as e
: # asyncio.TimeoutError
3104 if isinstance(e
, asyncio
.TimeoutError
):
3108 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3110 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3112 return 'FAILED', str(e
)
3114 return 'COMPLETED', output
3116 except (LcmException
, asyncio
.CancelledError
):
3118 except Exception as e
:
3119 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3121 async def action(self
, nsr_id
, nslcmop_id
):
3123 # Try to lock HA task here
3124 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3125 if not task_is_locked_by_me
:
3128 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3129 self
.logger
.debug(logging_text
+ "Enter")
3130 # get all needed from database
3134 db_nslcmop_update
= {}
3135 nslcmop_operation_state
= None
3136 error_description_nslcmop
= None
3139 # wait for any previous tasks in process
3140 step
= "Waiting for previous operations to terminate"
3141 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3143 self
._write
_ns
_status
(
3146 current_operation
="RUNNING ACTION",
3147 current_operation_id
=nslcmop_id
3150 step
= "Getting information from database"
3151 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3152 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3154 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3155 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3156 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3157 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3158 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3159 primitive
= db_nslcmop
["operationParams"]["primitive"]
3160 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3161 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3164 step
= "Getting vnfr from database"
3165 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3166 step
= "Getting vnfd from database"
3167 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3169 step
= "Getting nsd from database"
3170 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3172 # for backward compatibility
3173 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3174 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3175 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3176 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3178 # look for primitive
3179 config_primitive_desc
= None
3181 for vdu
in get_iterable(db_vnfd
, "vdu"):
3182 if vdu_id
== vdu
["id"]:
3183 for config_primitive
in deep_get(vdu
, ("vdu-configuration", "config-primitive"), ()):
3184 if config_primitive
["name"] == primitive
:
3185 config_primitive_desc
= config_primitive
3189 for kdu
in get_iterable(db_vnfd
, "kdu"):
3190 if kdu_name
== kdu
["name"]:
3191 for config_primitive
in deep_get(kdu
, ("kdu-configuration", "config-primitive"), ()):
3192 if config_primitive
["name"] == primitive
:
3193 config_primitive_desc
= config_primitive
3197 for config_primitive
in deep_get(db_vnfd
, ("vnf-configuration", "config-primitive"), ()):
3198 if config_primitive
["name"] == primitive
:
3199 config_primitive_desc
= config_primitive
3202 for config_primitive
in deep_get(db_nsd
, ("ns-configuration", "config-primitive"), ()):
3203 if config_primitive
["name"] == primitive
:
3204 config_primitive_desc
= config_primitive
3207 if not config_primitive_desc
and not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3208 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3213 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3214 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3216 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3217 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3219 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3221 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3224 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3226 # TODO check if ns is in a proper status
3227 if kdu_name
and (primitive
in ("upgrade", "rollback", "status") or kdu_action
):
3228 # kdur and desc_params already set from before
3229 if primitive_params
:
3230 desc_params
.update(primitive_params
)
3231 # TODO Check if we will need something at vnf level
3232 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3233 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3236 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3238 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3239 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3240 raise LcmException(msg
)
3242 db_dict
= {"collection": "nsrs",
3243 "filter": {"_id": nsr_id
},
3244 "path": "_admin.deployed.K8s.{}".format(index
)}
3245 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive
, vnf_index
, kdu_name
))
3246 step
= "Executing kdu {}".format(primitive
)
3247 if primitive
== "upgrade":
3248 if desc_params
.get("kdu_model"):
3249 kdu_model
= desc_params
.get("kdu_model")
3250 del desc_params
["kdu_model"]
3252 kdu_model
= kdu
.get("kdu-model")
3253 parts
= kdu_model
.split(sep
=":")
3255 kdu_model
= parts
[0]
3257 detailed_status
= await asyncio
.wait_for(
3258 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3259 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3260 kdu_instance
=kdu
.get("kdu-instance"),
3261 atomic
=True, kdu_model
=kdu_model
,
3262 params
=desc_params
, db_dict
=db_dict
,
3263 timeout
=timeout_ns_action
),
3264 timeout
=timeout_ns_action
+ 10)
3265 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3266 elif primitive
== "rollback":
3267 detailed_status
= await asyncio
.wait_for(
3268 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3269 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3270 kdu_instance
=kdu
.get("kdu-instance"),
3272 timeout
=timeout_ns_action
)
3273 elif primitive
== "status":
3274 detailed_status
= await asyncio
.wait_for(
3275 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3276 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3277 kdu_instance
=kdu
.get("kdu-instance")),
3278 timeout
=timeout_ns_action
)
3280 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3281 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3283 detailed_status
= await asyncio
.wait_for(
3284 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3285 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3286 kdu_instance
=kdu_instance
,
3287 primitive_name
=primitive
,
3288 params
=params
, db_dict
=db_dict
,
3289 timeout
=timeout_ns_action
),
3290 timeout
=timeout_ns_action
)
3293 nslcmop_operation_state
= 'COMPLETED'
3295 detailed_status
= ''
3296 nslcmop_operation_state
= 'FAILED'
3298 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3299 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3300 member_vnf_index
=vnf_index
,
3302 vdu_count_index
=vdu_count_index
),
3303 primitive
=primitive
,
3304 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3305 timeout
=timeout_ns_action
)
3307 db_nslcmop_update
["detailed-status"] = detailed_status
3308 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3309 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3311 return # database update is called inside finally
3313 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3314 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3316 except asyncio
.CancelledError
:
3317 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3318 exc
= "Operation was cancelled"
3319 except asyncio
.TimeoutError
:
3320 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3322 except Exception as e
:
3323 exc
= traceback
.format_exc()
3324 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3327 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3328 "FAILED {}: {}".format(step
, exc
)
3329 nslcmop_operation_state
= "FAILED"
3331 self
._write
_ns
_status
(
3333 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3334 current_operation
="IDLE",
3335 current_operation_id
=None,
3336 # error_description=error_description_nsr,
3337 # error_detail=error_detail,
3338 other_update
=db_nsr_update
3341 self
._write
_op
_status
(
3344 error_message
=error_description_nslcmop
,
3345 operation_state
=nslcmop_operation_state
,
3346 other_update
=db_nslcmop_update
,
3349 if nslcmop_operation_state
:
3351 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3352 "operationState": nslcmop_operation_state
},
3354 except Exception as e
:
3355 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3356 self
.logger
.debug(logging_text
+ "Exit")
3357 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3358 return nslcmop_operation_state
, detailed_status
3360 async def scale(self
, nsr_id
, nslcmop_id
):
3362 # Try to lock HA task here
3363 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3364 if not task_is_locked_by_me
:
3367 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3368 self
.logger
.debug(logging_text
+ "Enter")
3369 # get all needed from database
3372 db_nslcmop_update
= {}
3373 nslcmop_operation_state
= None
3376 # in case of error, indicates what part of scale was failed to put nsr at error status
3377 scale_process
= None
3378 old_operational_status
= ""
3379 old_config_status
= ""
3382 # wait for any previous tasks in process
3383 step
= "Waiting for previous operations to terminate"
3384 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3386 self
._write
_ns
_status
(
3389 current_operation
="SCALING",
3390 current_operation_id
=nslcmop_id
3393 step
= "Getting nslcmop from database"
3394 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3395 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3396 step
= "Getting nsr from database"
3397 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3399 old_operational_status
= db_nsr
["operational-status"]
3400 old_config_status
= db_nsr
["config-status"]
3401 step
= "Parsing scaling parameters"
3402 # self.logger.debug(step)
3403 db_nsr_update
["operational-status"] = "scaling"
3404 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3405 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3408 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3409 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3410 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3411 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3412 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3415 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3416 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3417 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3418 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3419 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3421 # for backward compatibility
3422 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3423 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3424 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3425 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3427 step
= "Getting vnfr from database"
3428 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3429 step
= "Getting vnfd from database"
3430 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3432 step
= "Getting scaling-group-descriptor"
3433 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3434 if scaling_descriptor
["name"] == scaling_group
:
3437 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3438 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3441 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3442 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3443 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3446 # TODO check if ns is in a proper status
3447 step
= "Sending scale order to VIM"
3449 if not db_nsr
["_admin"].get("scaling-group"):
3450 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3451 admin_scale_index
= 0
3453 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3454 if admin_scale_info
["name"] == scaling_group
:
3455 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3457 else: # not found, set index one plus last element and add new entry with the name
3458 admin_scale_index
+= 1
3459 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3460 RO_scaling_info
= []
3461 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3462 if scaling_type
== "SCALE_OUT":
3463 # count if max-instance-count is reached
3464 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3465 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3466 if nb_scale_op
>= max_instance_count
:
3467 raise LcmException("reached the limit of {} (max-instance-count) "
3468 "scaling-out operations for the "
3469 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3472 vdu_scaling_info
["scaling_direction"] = "OUT"
3473 vdu_scaling_info
["vdu-create"] = {}
3474 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3475 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3476 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3477 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3479 elif scaling_type
== "SCALE_IN":
3480 # count if min-instance-count is reached
3481 min_instance_count
= 0
3482 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3483 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3484 if nb_scale_op
<= min_instance_count
:
3485 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3486 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3488 vdu_scaling_info
["scaling_direction"] = "IN"
3489 vdu_scaling_info
["vdu-delete"] = {}
3490 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3491 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3492 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3493 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3495 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3496 vdu_create
= vdu_scaling_info
.get("vdu-create")
3497 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3498 if vdu_scaling_info
["scaling_direction"] == "IN":
3499 for vdur
in reversed(db_vnfr
["vdur"]):
3500 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3501 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3502 vdu_scaling_info
["vdu"].append({
3503 "name": vdur
["name"],
3504 "vdu_id": vdur
["vdu-id-ref"],
3507 for interface
in vdur
["interfaces"]:
3508 vdu_scaling_info
["vdu"][-1]["interface"].append({
3509 "name": interface
["name"],
3510 "ip_address": interface
["ip-address"],
3511 "mac_address": interface
.get("mac-address"),
3513 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
3516 step
= "Executing pre-scale vnf-config-primitive"
3517 if scaling_descriptor
.get("scaling-config-action"):
3518 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3519 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
3520 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
3521 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3522 step
= db_nslcmop_update
["detailed-status"] = \
3523 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3525 # look for primitive
3526 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3527 if config_primitive
["name"] == vnf_config_primitive
:
3531 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
3532 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
3533 "primitive".format(scaling_group
, config_primitive
))
3535 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3536 if db_vnfr
.get("additionalParamsForVnf"):
3537 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3539 scale_process
= "VCA"
3540 db_nsr_update
["config-status"] = "configuring pre-scaling"
3541 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3543 # Pre-scale reintent check: Check if this sub-operation has been executed before
3544 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3545 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
3546 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3547 # Skip sub-operation
3548 result
= 'COMPLETED'
3549 result_detail
= 'Done'
3550 self
.logger
.debug(logging_text
+
3551 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
3552 vnf_config_primitive
, result
, result_detail
))
3554 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3555 # New sub-operation: Get index of this sub-operation
3556 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3557 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3558 format(vnf_config_primitive
))
3560 # Reintent: Get registered params for this existing sub-operation
3561 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3562 vnf_index
= op
.get('member_vnf_index')
3563 vnf_config_primitive
= op
.get('primitive')
3564 primitive_params
= op
.get('primitive_params')
3565 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3566 format(vnf_config_primitive
))
3567 # Execute the primitive, either with new (first-time) or registered (reintent) args
3568 result
, result_detail
= await self
._ns
_execute
_primitive
(
3569 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3570 member_vnf_index
=vnf_index
,
3572 vdu_count_index
=None),
3573 vnf_config_primitive
, primitive_params
)
3574 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3575 vnf_config_primitive
, result
, result_detail
))
3576 # Update operationState = COMPLETED | FAILED
3577 self
._update
_suboperation
_status
(
3578 db_nslcmop
, op_index
, result
, result_detail
)
3580 if result
== "FAILED":
3581 raise LcmException(result_detail
)
3582 db_nsr_update
["config-status"] = old_config_status
3583 scale_process
= None
3587 # Should this block be skipped if 'RO_nsr_id' == None ?
3588 # if (RO_nsr_id and RO_scaling_info):
3590 scale_process
= "RO"
3591 # Scale RO reintent check: Check if this sub-operation has been executed before
3592 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3593 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
3594 if (op_index
== self
.SUBOPERATION_STATUS_SKIP
):
3595 # Skip sub-operation
3596 result
= 'COMPLETED'
3597 result_detail
= 'Done'
3598 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
3599 result
, result_detail
))
3601 if (op_index
== self
.SUBOPERATION_STATUS_NEW
):
3602 # New sub-operation: Get index of this sub-operation
3603 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3604 self
.logger
.debug(logging_text
+ "New sub-operation RO")
3606 # Reintent: Get registered params for this existing sub-operation
3607 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3608 RO_nsr_id
= op
.get('RO_nsr_id')
3609 RO_scaling_info
= op
.get('RO_scaling_info')
3610 self
.logger
.debug(logging_text
+ "Sub-operation RO reintent".format(
3611 vnf_config_primitive
))
3613 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
3614 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
3615 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
3617 RO_nslcmop_id
= RO_desc
["instance_action_id"]
3618 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
3620 RO_task_done
= False
3621 step
= detailed_status
= "Waiting RO_task_id={} to complete the scale action.".format(RO_nslcmop_id
)
3622 detailed_status_old
= None
3623 self
.logger
.debug(logging_text
+ step
)
3625 deployment_timeout
= 1 * 3600 # One hour
3626 while deployment_timeout
> 0:
3627 if not RO_task_done
:
3628 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
3629 extra_item_id
=RO_nslcmop_id
)
3632 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3634 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3635 if ns_status
== "ERROR":
3636 raise ROclient
.ROClientException(ns_status_info
)
3637 elif ns_status
== "BUILD":
3638 detailed_status
= step
+ "; {}".format(ns_status_info
)
3639 elif ns_status
== "ACTIVE":
3641 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
3642 self
.logger
.debug(logging_text
+ step
)
3644 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3647 if ns_status
== "ERROR":
3648 raise ROclient
.ROClientException(ns_status_info
)
3649 elif ns_status
== "BUILD":
3650 detailed_status
= step
+ "; {}".format(ns_status_info
)
3651 elif ns_status
== "ACTIVE":
3652 step
= detailed_status
= \
3653 "Waiting for management IP address reported by the VIM. Updating VNFRs"
3655 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
3658 desc
= await self
.RO
.show("ns", RO_nsr_id
)
3661 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3663 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
3664 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
3666 except LcmExceptionNoMgmtIP
:
3669 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
3670 if detailed_status
!= detailed_status_old
:
3671 self
._update
_suboperation
_status
(
3672 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
3673 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
3674 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3676 await asyncio
.sleep(5, loop
=self
.loop
)
3677 deployment_timeout
-= 5
3678 if deployment_timeout
<= 0:
3679 self
._update
_suboperation
_status
(
3680 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
3681 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
3683 # update VDU_SCALING_INFO with the obtained ip_addresses
3684 if vdu_scaling_info
["scaling_direction"] == "OUT":
3685 for vdur
in reversed(db_vnfr
["vdur"]):
3686 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
3687 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
3688 vdu_scaling_info
["vdu"].append({
3689 "name": vdur
["name"],
3690 "vdu_id": vdur
["vdu-id-ref"],
3693 for interface
in vdur
["interfaces"]:
3694 vdu_scaling_info
["vdu"][-1]["interface"].append({
3695 "name": interface
["name"],
3696 "ip_address": interface
["ip-address"],
3697 "mac_address": interface
.get("mac-address"),
3699 del vdu_scaling_info
["vdu-create"]
3701 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
3704 scale_process
= None
3706 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3709 # execute primitive service POST-SCALING
3710 step
= "Executing post-scale vnf-config-primitive"
3711 if scaling_descriptor
.get("scaling-config-action"):
3712 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3713 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
3714 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
3715 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3716 step
= db_nslcmop_update
["detailed-status"] = \
3717 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3719 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3720 if db_vnfr
.get("additionalParamsForVnf"):
3721 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3723 # look for primitive
3724 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3725 if config_primitive
["name"] == vnf_config_primitive
:
3728 raise LcmException("Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:"
3729 "scaling-config-action[vnf-config-primitive-name-ref='{}'] does not "
3730 "match any vnf-configuration:config-primitive".format(scaling_group
,
3732 scale_process
= "VCA"
3733 db_nsr_update
["config-status"] = "configuring post-scaling"
3734 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3736 # Post-scale reintent check: Check if this sub-operation has been executed before
3737 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3738 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
3739 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3740 # Skip sub-operation
3741 result
= 'COMPLETED'
3742 result_detail
= 'Done'
3743 self
.logger
.debug(logging_text
+
3744 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
3745 format(vnf_config_primitive
, result
, result_detail
))
3747 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3748 # New sub-operation: Get index of this sub-operation
3749 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3750 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3751 format(vnf_config_primitive
))
3753 # Reintent: Get registered params for this existing sub-operation
3754 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3755 vnf_index
= op
.get('member_vnf_index')
3756 vnf_config_primitive
= op
.get('primitive')
3757 primitive_params
= op
.get('primitive_params')
3758 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation reintent".
3759 format(vnf_config_primitive
))
3760 # Execute the primitive, either with new (first-time) or registered (reintent) args
3761 result
, result_detail
= await self
._ns
_execute
_primitive
(
3762 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3763 member_vnf_index
=vnf_index
,
3765 vdu_count_index
=None),
3766 vnf_config_primitive
, primitive_params
)
3767 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3768 vnf_config_primitive
, result
, result_detail
))
3769 # Update operationState = COMPLETED | FAILED
3770 self
._update
_suboperation
_status
(
3771 db_nslcmop
, op_index
, result
, result_detail
)
3773 if result
== "FAILED":
3774 raise LcmException(result_detail
)
3775 db_nsr_update
["config-status"] = old_config_status
3776 scale_process
= None
3779 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
3780 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
3781 else old_operational_status
3782 db_nsr_update
["config-status"] = old_config_status
3784 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
3785 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3787 except asyncio
.CancelledError
:
3788 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3789 exc
= "Operation was cancelled"
3790 except Exception as e
:
3791 exc
= traceback
.format_exc()
3792 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3794 self
._write
_ns
_status
(
3797 current_operation
="IDLE",
3798 current_operation_id
=None
3801 db_nslcmop_update
["detailed-status"] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
3802 nslcmop_operation_state
= "FAILED"
3804 db_nsr_update
["operational-status"] = old_operational_status
3805 db_nsr_update
["config-status"] = old_config_status
3806 db_nsr_update
["detailed-status"] = ""
3808 if "VCA" in scale_process
:
3809 db_nsr_update
["config-status"] = "failed"
3810 if "RO" in scale_process
:
3811 db_nsr_update
["operational-status"] = "failed"
3812 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
3815 error_description_nslcmop
= None
3816 nslcmop_operation_state
= "COMPLETED"
3817 db_nslcmop_update
["detailed-status"] = "Done"
3819 self
._write
_op
_status
(
3822 error_message
=error_description_nslcmop
,
3823 operation_state
=nslcmop_operation_state
,
3824 other_update
=db_nslcmop_update
,
3827 self
._write
_ns
_status
(
3830 current_operation
="IDLE",
3831 current_operation_id
=None,
3832 other_update
=db_nsr_update
3835 if nslcmop_operation_state
:
3837 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3838 "operationState": nslcmop_operation_state
},
3841 # await asyncio.sleep(cooldown_time, loop=self.loop)
3842 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
3843 except Exception as e
:
3844 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3845 self
.logger
.debug(logging_text
+ "Exit")
3846 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")