1 # -*- coding: utf-8 -*-
4 # Copyright 2018 Telefonica S.A.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
22 import logging
.handlers
25 from jinja2
import Environment
, Template
, meta
, TemplateError
, TemplateNotFound
, TemplateSyntaxError
27 from osm_lcm
import ROclient
28 from osm_lcm
.ng_ro
import NgRoClient
, NgRoException
29 from osm_lcm
.lcm_utils
import LcmException
, LcmExceptionNoMgmtIP
, LcmBase
, deep_get
, get_iterable
, populate_dict
30 from n2vc
.k8s_helm_conn
import K8sHelmConnector
31 from n2vc
.k8s_juju_conn
import K8sJujuConnector
33 from osm_common
.dbbase
import DbException
34 from osm_common
.fsbase
import FsException
36 from n2vc
.n2vc_juju_conn
import N2VCJujuConnector
37 from n2vc
.exceptions
import N2VCException
, N2VCNotFound
, K8sException
39 from copy
import copy
, deepcopy
40 from http
import HTTPStatus
42 from uuid
import uuid4
43 from functools
import partial
45 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
49 timeout_vca_on_error
= 5 * 60 # Time for charm from first time at blocked,error status to mark as failed
50 timeout_ns_deploy
= 2 * 3600 # default global timeout for deployment a ns
51 timeout_ns_terminate
= 1800 # default global timeout for un deployment a ns
52 timeout_charm_delete
= 10 * 60
53 timeout_primitive
= 30 * 60 # timeout for primitive execution
54 timeout_progress_primitive
= 10 * 60 # timeout for some progress in a primitive execution
56 SUBOPERATION_STATUS_NOT_FOUND
= -1
57 SUBOPERATION_STATUS_NEW
= -2
58 SUBOPERATION_STATUS_SKIP
= -3
59 task_name_deploy_vca
= "Deploying VCA"
61 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
63 Init, Connect to database, filesystem storage, and messaging
64 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
71 logger
=logging
.getLogger('lcm.ns')
75 self
.lcm_tasks
= lcm_tasks
76 self
.timeout
= config
["timeout"]
77 self
.ro_config
= config
["ro_config"]
78 self
.ng_ro
= config
["ro_config"].get("ng")
79 self
.vca_config
= config
["VCA"].copy()
81 # create N2VC connector
82 self
.n2vc
= N2VCJujuConnector(
87 url
='{}:{}'.format(self
.vca_config
['host'], self
.vca_config
['port']),
88 username
=self
.vca_config
.get('user', None),
89 vca_config
=self
.vca_config
,
90 on_update_db
=self
._on
_update
_n
2vc
_db
93 self
.k8sclusterhelm
= K8sHelmConnector(
94 kubectl_command
=self
.vca_config
.get("kubectlpath"),
95 helm_command
=self
.vca_config
.get("helmpath"),
102 self
.k8sclusterjuju
= K8sJujuConnector(
103 kubectl_command
=self
.vca_config
.get("kubectlpath"),
104 juju_command
=self
.vca_config
.get("jujupath"),
111 self
.k8scluster_map
= {
112 "helm-chart": self
.k8sclusterhelm
,
113 "chart": self
.k8sclusterhelm
,
114 "juju-bundle": self
.k8sclusterjuju
,
115 "juju": self
.k8sclusterjuju
,
119 self
.RO
= NgRoClient(self
.loop
, **self
.ro_config
)
121 self
.RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
123 def _on_update_ro_db(self
, nsrs_id
, ro_descriptor
):
125 # self.logger.debug('_on_update_ro_db(nsrs_id={}'.format(nsrs_id))
128 # TODO filter RO descriptor fields...
132 # db_dict['deploymentStatus'] = yaml.dump(ro_descriptor, default_flow_style=False, indent=2)
133 db_dict
['deploymentStatus'] = ro_descriptor
134 self
.update_db_2("nsrs", nsrs_id
, db_dict
)
136 except Exception as e
:
137 self
.logger
.warn('Cannot write database RO deployment for ns={} -> {}'.format(nsrs_id
, e
))
139 async def _on_update_n2vc_db(self
, table
, filter, path
, updated_data
):
141 # remove last dot from path (if exists)
142 if path
.endswith('.'):
145 # self.logger.debug('_on_update_n2vc_db(table={}, filter={}, path={}, updated_data={}'
146 # .format(table, filter, path, updated_data))
150 nsr_id
= filter.get('_id')
152 # read ns record from database
153 nsr
= self
.db
.get_one(table
='nsrs', q_filter
=filter)
154 current_ns_status
= nsr
.get('nsState')
156 # get vca status for NS
157 status_dict
= await self
.n2vc
.get_status(namespace
='.' + nsr_id
, yaml_format
=False)
161 db_dict
['vcaStatus'] = status_dict
163 # update configurationStatus for this VCA
165 vca_index
= int(path
[path
.rfind(".")+1:])
167 vca_list
= deep_get(target_dict
=nsr
, key_list
=('_admin', 'deployed', 'VCA'))
168 vca_status
= vca_list
[vca_index
].get('status')
170 configuration_status_list
= nsr
.get('configurationStatus')
171 config_status
= configuration_status_list
[vca_index
].get('status')
173 if config_status
== 'BROKEN' and vca_status
!= 'failed':
174 db_dict
['configurationStatus'][vca_index
] = 'READY'
175 elif config_status
!= 'BROKEN' and vca_status
== 'failed':
176 db_dict
['configurationStatus'][vca_index
] = 'BROKEN'
177 except Exception as e
:
178 # not update configurationStatus
179 self
.logger
.debug('Error updating vca_index (ignore): {}'.format(e
))
181 # if nsState = 'READY' check if juju is reporting some error => nsState = 'DEGRADED'
182 # if nsState = 'DEGRADED' check if all is OK
184 if current_ns_status
in ('READY', 'DEGRADED'):
185 error_description
= ''
187 if status_dict
.get('machines'):
188 for machine_id
in status_dict
.get('machines'):
189 machine
= status_dict
.get('machines').get(machine_id
)
190 # check machine agent-status
191 if machine
.get('agent-status'):
192 s
= machine
.get('agent-status').get('status')
195 error_description
+= 'machine {} agent-status={} ; '.format(machine_id
, s
)
196 # check machine instance status
197 if machine
.get('instance-status'):
198 s
= machine
.get('instance-status').get('status')
201 error_description
+= 'machine {} instance-status={} ; '.format(machine_id
, s
)
203 if status_dict
.get('applications'):
204 for app_id
in status_dict
.get('applications'):
205 app
= status_dict
.get('applications').get(app_id
)
206 # check application status
207 if app
.get('status'):
208 s
= app
.get('status').get('status')
211 error_description
+= 'application {} status={} ; '.format(app_id
, s
)
213 if error_description
:
214 db_dict
['errorDescription'] = error_description
215 if current_ns_status
== 'READY' and is_degraded
:
216 db_dict
['nsState'] = 'DEGRADED'
217 if current_ns_status
== 'DEGRADED' and not is_degraded
:
218 db_dict
['nsState'] = 'READY'
221 self
.update_db_2("nsrs", nsr_id
, db_dict
)
223 except (asyncio
.CancelledError
, asyncio
.TimeoutError
):
225 except Exception as e
:
226 self
.logger
.warn('Error updating NS state for ns={}: {}'.format(nsr_id
, e
))
228 def vnfd2RO(self
, vnfd
, new_id
=None, additionalParams
=None, nsrId
=None):
230 Converts creates a new vnfd descriptor for RO base on input OSM IM vnfd
231 :param vnfd: input vnfd
232 :param new_id: overrides vnf id if provided
233 :param additionalParams: Instantiation params for VNFs provided
234 :param nsrId: Id of the NSR
235 :return: copy of vnfd
238 vnfd_RO
= deepcopy(vnfd
)
239 # remove unused by RO configuration, monitoring, scaling and internal keys
240 vnfd_RO
.pop("_id", None)
241 vnfd_RO
.pop("_admin", None)
242 vnfd_RO
.pop("vnf-configuration", None)
243 vnfd_RO
.pop("monitoring-param", None)
244 vnfd_RO
.pop("scaling-group-descriptor", None)
245 vnfd_RO
.pop("kdu", None)
246 vnfd_RO
.pop("k8s-cluster", None)
248 vnfd_RO
["id"] = new_id
250 # parse cloud-init or cloud-init-file with the provided variables using Jinja2
251 for vdu
in get_iterable(vnfd_RO
, "vdu"):
252 cloud_init_file
= None
253 if vdu
.get("cloud-init-file"):
254 base_folder
= vnfd
["_admin"]["storage"]
255 cloud_init_file
= "{}/{}/cloud_init/{}".format(base_folder
["folder"], base_folder
["pkg-dir"],
256 vdu
["cloud-init-file"])
257 with self
.fs
.file_open(cloud_init_file
, "r") as ci_file
:
258 cloud_init_content
= ci_file
.read()
259 vdu
.pop("cloud-init-file", None)
260 elif vdu
.get("cloud-init"):
261 cloud_init_content
= vdu
["cloud-init"]
266 ast
= env
.parse(cloud_init_content
)
267 mandatory_vars
= meta
.find_undeclared_variables(ast
)
269 for var
in mandatory_vars
:
270 if not additionalParams
or var
not in additionalParams
.keys():
271 raise LcmException("Variable '{}' defined at vnfd[id={}]:vdu[id={}]:cloud-init/cloud-init-"
272 "file, must be provided in the instantiation parameters inside the "
273 "'additionalParamsForVnf' block".format(var
, vnfd
["id"], vdu
["id"]))
274 template
= Template(cloud_init_content
)
275 cloud_init_content
= template
.render(additionalParams
or {})
276 vdu
["cloud-init"] = cloud_init_content
279 except FsException
as e
:
280 raise LcmException("Error reading vnfd[id={}]:vdu[id={}]:cloud-init-file={}: {}".
281 format(vnfd
["id"], vdu
["id"], cloud_init_file
, e
))
282 except (TemplateError
, TemplateNotFound
, TemplateSyntaxError
) as e
:
283 raise LcmException("Error parsing Jinja2 to cloud-init content at vnfd[id={}]:vdu[id={}]: {}".
284 format(vnfd
["id"], vdu
["id"], e
))
286 def _ns_params_2_RO(self
, ns_params
, nsd
, vnfd_dict
, db_vnfrs
, n2vc_key_list
):
288 Creates a RO ns descriptor from OSM ns_instantiate params
289 :param ns_params: OSM instantiate params
290 :param vnfd_dict: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
291 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index. {member-vnf-index: {vnfr_object}, ...}
292 :return: The RO ns descriptor
296 # TODO feature 1417: Check that no instantiation is set over PDU
297 # check if PDU forces a concrete vim-network-id and add it
298 # check if PDU contains a SDN-assist info (dpid, switch, port) and pass it to RO
300 def vim_account_2_RO(vim_account
):
301 if vim_account
in vim_2_RO
:
302 return vim_2_RO
[vim_account
]
304 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_account
})
305 if db_vim
["_admin"]["operationalState"] != "ENABLED":
306 raise LcmException("VIM={} is not available. operationalState={}".format(
307 vim_account
, db_vim
["_admin"]["operationalState"]))
308 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
309 vim_2_RO
[vim_account
] = RO_vim_id
312 def wim_account_2_RO(wim_account
):
313 if isinstance(wim_account
, str):
314 if wim_account
in wim_2_RO
:
315 return wim_2_RO
[wim_account
]
317 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_account
})
318 if db_wim
["_admin"]["operationalState"] != "ENABLED":
319 raise LcmException("WIM={} is not available. operationalState={}".format(
320 wim_account
, db_wim
["_admin"]["operationalState"]))
321 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO-account"]
322 wim_2_RO
[wim_account
] = RO_wim_id
327 def ip_profile_2_RO(ip_profile
):
328 RO_ip_profile
= deepcopy((ip_profile
))
329 if "dns-server" in RO_ip_profile
:
330 if isinstance(RO_ip_profile
["dns-server"], list):
331 RO_ip_profile
["dns-address"] = []
332 for ds
in RO_ip_profile
.pop("dns-server"):
333 RO_ip_profile
["dns-address"].append(ds
['address'])
335 RO_ip_profile
["dns-address"] = RO_ip_profile
.pop("dns-server")
336 if RO_ip_profile
.get("ip-version") == "ipv4":
337 RO_ip_profile
["ip-version"] = "IPv4"
338 if RO_ip_profile
.get("ip-version") == "ipv6":
339 RO_ip_profile
["ip-version"] = "IPv6"
340 if "dhcp-params" in RO_ip_profile
:
341 RO_ip_profile
["dhcp"] = RO_ip_profile
.pop("dhcp-params")
347 # "name": ns_params["nsName"],
348 # "description": ns_params.get("nsDescription"),
349 "datacenter": vim_account_2_RO(ns_params
["vimAccountId"]),
350 "wim_account": wim_account_2_RO(ns_params
.get("wimAccountId")),
351 # "scenario": ns_params["nsdId"],
353 # set vim_account of each vnf if different from general vim_account.
354 # Get this information from <vnfr> database content, key vim-account-id
355 # Vim account can be set by placement_engine and it may be different from
356 # the instantiate parameters (vnfs.member-vnf-index.datacenter).
357 for vnf_index
, vnfr
in db_vnfrs
.items():
358 if vnfr
.get("vim-account-id") and vnfr
["vim-account-id"] != ns_params
["vimAccountId"]:
359 populate_dict(RO_ns_params
, ("vnfs", vnf_index
, "datacenter"), vim_account_2_RO(vnfr
["vim-account-id"]))
361 n2vc_key_list
= n2vc_key_list
or []
362 for vnfd_ref
, vnfd
in vnfd_dict
.items():
363 vdu_needed_access
= []
365 if vnfd
.get("vnf-configuration"):
366 ssh_required
= deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required"))
367 if ssh_required
and vnfd
.get("mgmt-interface"):
368 if vnfd
["mgmt-interface"].get("vdu-id"):
369 vdu_needed_access
.append(vnfd
["mgmt-interface"]["vdu-id"])
370 elif vnfd
["mgmt-interface"].get("cp"):
371 mgmt_cp
= vnfd
["mgmt-interface"]["cp"]
373 for vdu
in vnfd
.get("vdu", ()):
374 if vdu
.get("vdu-configuration"):
375 ssh_required
= deep_get(vdu
, ("vdu-configuration", "config-access", "ssh-access", "required"))
377 vdu_needed_access
.append(vdu
["id"])
379 for vdu_interface
in vdu
.get("interface"):
380 if vdu_interface
.get("external-connection-point-ref") and \
381 vdu_interface
["external-connection-point-ref"] == mgmt_cp
:
382 vdu_needed_access
.append(vdu
["id"])
386 if vdu_needed_access
:
387 for vnf_member
in nsd
.get("constituent-vnfd"):
388 if vnf_member
["vnfd-id-ref"] != vnfd_ref
:
390 for vdu
in vdu_needed_access
:
391 populate_dict(RO_ns_params
,
392 ("vnfs", vnf_member
["member-vnf-index"], "vdus", vdu
, "mgmt_keys"),
395 if ns_params
.get("vduImage"):
396 RO_ns_params
["vduImage"] = ns_params
["vduImage"]
398 if ns_params
.get("ssh_keys"):
399 RO_ns_params
["cloud-config"] = {"key-pairs": ns_params
["ssh_keys"]}
400 for vnf_params
in get_iterable(ns_params
, "vnf"):
401 for constituent_vnfd
in nsd
["constituent-vnfd"]:
402 if constituent_vnfd
["member-vnf-index"] == vnf_params
["member-vnf-index"]:
403 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
406 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index={} is not present at nsd:"
407 "constituent-vnfd".format(vnf_params
["member-vnf-index"]))
409 for vdu_params
in get_iterable(vnf_params
, "vdu"):
410 # TODO feature 1417: check that this VDU exist and it is not a PDU
411 if vdu_params
.get("volume"):
412 for volume_params
in vdu_params
["volume"]:
413 if volume_params
.get("vim-volume-id"):
414 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
415 vdu_params
["id"], "devices", volume_params
["name"], "vim_id"),
416 volume_params
["vim-volume-id"])
417 if vdu_params
.get("interface"):
418 for interface_params
in vdu_params
["interface"]:
419 if interface_params
.get("ip-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
["ip-address"])
424 if interface_params
.get("mac-address"):
425 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
426 vdu_params
["id"], "interfaces", interface_params
["name"],
428 interface_params
["mac-address"])
429 if interface_params
.get("floating-ip-required"):
430 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
431 vdu_params
["id"], "interfaces", interface_params
["name"],
433 interface_params
["floating-ip-required"])
435 for internal_vld_params
in get_iterable(vnf_params
, "internal-vld"):
436 if internal_vld_params
.get("vim-network-name"):
437 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
438 internal_vld_params
["name"], "vim-network-name"),
439 internal_vld_params
["vim-network-name"])
440 if internal_vld_params
.get("vim-network-id"):
441 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
442 internal_vld_params
["name"], "vim-network-id"),
443 internal_vld_params
["vim-network-id"])
444 if internal_vld_params
.get("ip-profile"):
445 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
446 internal_vld_params
["name"], "ip-profile"),
447 ip_profile_2_RO(internal_vld_params
["ip-profile"]))
448 if internal_vld_params
.get("provider-network"):
450 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "networks",
451 internal_vld_params
["name"], "provider-network"),
452 internal_vld_params
["provider-network"].copy())
454 for icp_params
in get_iterable(internal_vld_params
, "internal-connection-point"):
457 for vdu_descriptor
in vnf_descriptor
["vdu"]:
458 for vdu_interface
in vdu_descriptor
["interface"]:
459 if vdu_interface
.get("internal-connection-point-ref") == icp_params
["id-ref"]:
460 if icp_params
.get("ip-address"):
461 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
462 vdu_descriptor
["id"], "interfaces",
463 vdu_interface
["name"], "ip_address"),
464 icp_params
["ip-address"])
466 if icp_params
.get("mac-address"):
467 populate_dict(RO_ns_params
, ("vnfs", vnf_params
["member-vnf-index"], "vdus",
468 vdu_descriptor
["id"], "interfaces",
469 vdu_interface
["name"], "mac_address"),
470 icp_params
["mac-address"])
476 raise LcmException("Invalid instantiate parameter vnf:member-vnf-index[{}]:"
477 "internal-vld:id-ref={} is not present at vnfd:internal-"
478 "connection-point".format(vnf_params
["member-vnf-index"],
479 icp_params
["id-ref"]))
481 for vld_params
in get_iterable(ns_params
, "vld"):
482 if "ip-profile" in vld_params
:
483 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "ip-profile"),
484 ip_profile_2_RO(vld_params
["ip-profile"]))
486 if vld_params
.get("provider-network"):
488 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "provider-network"),
489 vld_params
["provider-network"].copy())
491 if "wimAccountId" in vld_params
and vld_params
["wimAccountId"] is not None:
492 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "wim_account"),
493 wim_account_2_RO(vld_params
["wimAccountId"])),
494 if vld_params
.get("vim-network-name"):
496 if isinstance(vld_params
["vim-network-name"], dict):
497 for vim_account
, vim_net
in vld_params
["vim-network-name"].items():
498 RO_vld_sites
.append({
499 "netmap-use": vim_net
,
500 "datacenter": vim_account_2_RO(vim_account
)
502 else: # isinstance str
503 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-name"]})
505 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
507 if vld_params
.get("vim-network-id"):
509 if isinstance(vld_params
["vim-network-id"], dict):
510 for vim_account
, vim_net
in vld_params
["vim-network-id"].items():
511 RO_vld_sites
.append({
512 "netmap-use": vim_net
,
513 "datacenter": vim_account_2_RO(vim_account
)
515 else: # isinstance str
516 RO_vld_sites
.append({"netmap-use": vld_params
["vim-network-id"]})
518 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "sites"), RO_vld_sites
)
519 if vld_params
.get("ns-net"):
520 if isinstance(vld_params
["ns-net"], dict):
521 for vld_id
, instance_scenario_id
in vld_params
["ns-net"].items():
522 RO_vld_ns_net
= {"instance_scenario_id": instance_scenario_id
, "osm_id": vld_id
}
523 populate_dict(RO_ns_params
, ("networks", vld_params
["name"], "use-network"), RO_vld_ns_net
)
524 if "vnfd-connection-point-ref" in vld_params
:
525 for cp_params
in vld_params
["vnfd-connection-point-ref"]:
527 for constituent_vnfd
in nsd
["constituent-vnfd"]:
528 if constituent_vnfd
["member-vnf-index"] == cp_params
["member-vnf-index-ref"]:
529 vnf_descriptor
= vnfd_dict
[constituent_vnfd
["vnfd-id-ref"]]
533 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={} "
534 "is not present at nsd:constituent-vnfd".format(cp_params
["member-vnf-index-ref"]))
536 for vdu_descriptor
in vnf_descriptor
["vdu"]:
537 for interface_descriptor
in vdu_descriptor
["interface"]:
538 if interface_descriptor
.get("external-connection-point-ref") == \
539 cp_params
["vnfd-connection-point-ref"]:
546 "Invalid instantiate parameter vld:vnfd-connection-point-ref:member-vnf-index-ref={}:"
547 "vnfd-connection-point-ref={} is not present at vnfd={}".format(
548 cp_params
["member-vnf-index-ref"],
549 cp_params
["vnfd-connection-point-ref"],
550 vnf_descriptor
["id"]))
551 if cp_params
.get("ip-address"):
552 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
553 vdu_descriptor
["id"], "interfaces",
554 interface_descriptor
["name"], "ip_address"),
555 cp_params
["ip-address"])
556 if cp_params
.get("mac-address"):
557 populate_dict(RO_ns_params
, ("vnfs", cp_params
["member-vnf-index-ref"], "vdus",
558 vdu_descriptor
["id"], "interfaces",
559 interface_descriptor
["name"], "mac_address"),
560 cp_params
["mac-address"])
563 def scale_vnfr(self
, db_vnfr
, vdu_create
=None, vdu_delete
=None):
564 # make a copy to do not change
565 vdu_create
= copy(vdu_create
)
566 vdu_delete
= copy(vdu_delete
)
568 vdurs
= db_vnfr
.get("vdur")
571 vdu_index
= len(vdurs
)
574 vdur
= vdurs
[vdu_index
]
575 if vdur
.get("pdu-type"):
577 vdu_id_ref
= vdur
["vdu-id-ref"]
578 if vdu_create
and vdu_create
.get(vdu_id_ref
):
579 for index
in range(0, vdu_create
[vdu_id_ref
]):
580 vdur
= deepcopy(vdur
)
581 vdur
["_id"] = str(uuid4())
582 vdur
["count-index"] += 1
583 vdurs
.insert(vdu_index
+1+index
, vdur
)
584 del vdu_create
[vdu_id_ref
]
585 if vdu_delete
and vdu_delete
.get(vdu_id_ref
):
587 vdu_delete
[vdu_id_ref
] -= 1
588 if not vdu_delete
[vdu_id_ref
]:
589 del vdu_delete
[vdu_id_ref
]
590 # check all operations are done
591 if vdu_create
or vdu_delete
:
592 raise LcmException("Error scaling OUT VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
595 raise LcmException("Error scaling IN VNFR for {}. There is not any existing vnfr. Scaled to 0?".format(
598 vnfr_update
= {"vdur": vdurs
}
599 db_vnfr
["vdur"] = vdurs
600 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
602 def ns_update_nsr(self
, ns_update_nsr
, db_nsr
, nsr_desc_RO
):
604 Updates database nsr with the RO info for the created vld
605 :param ns_update_nsr: dictionary to be filled with the updated info
606 :param db_nsr: content of db_nsr. This is also modified
607 :param nsr_desc_RO: nsr descriptor from RO
608 :return: Nothing, LcmException is raised on errors
611 for vld_index
, vld
in enumerate(get_iterable(db_nsr
, "vld")):
612 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
613 if vld
["id"] != net_RO
.get("ns_net_osm_id"):
615 vld
["vim-id"] = net_RO
.get("vim_net_id")
616 vld
["name"] = net_RO
.get("vim_name")
617 vld
["status"] = net_RO
.get("status")
618 vld
["status-detailed"] = net_RO
.get("error_msg")
619 ns_update_nsr
["vld.{}".format(vld_index
)] = vld
622 raise LcmException("ns_update_nsr: Not found vld={} at RO info".format(vld
["id"]))
624 def set_vnfr_at_error(self
, db_vnfrs
, error_text
):
626 for db_vnfr
in db_vnfrs
.values():
627 vnfr_update
= {"status": "ERROR"}
628 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
629 if "status" not in vdur
:
630 vdur
["status"] = "ERROR"
631 vnfr_update
["vdur.{}.status".format(vdu_index
)] = "ERROR"
633 vdur
["status-detailed"] = str(error_text
)
634 vnfr_update
["vdur.{}.status-detailed".format(vdu_index
)] = "ERROR"
635 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
636 except DbException
as e
:
637 self
.logger
.error("Cannot update vnf. {}".format(e
))
639 def ns_update_vnfr(self
, db_vnfrs
, nsr_desc_RO
):
641 Updates database vnfr with the RO info, e.g. ip_address, vim_id... Descriptor db_vnfrs is also updated
642 :param db_vnfrs: dictionary with member-vnf-index: vnfr-content
643 :param nsr_desc_RO: nsr descriptor from RO
644 :return: Nothing, LcmException is raised on errors
646 for vnf_index
, db_vnfr
in db_vnfrs
.items():
647 for vnf_RO
in nsr_desc_RO
["vnfs"]:
648 if vnf_RO
["member_vnf_index"] != vnf_index
:
651 if vnf_RO
.get("ip_address"):
652 db_vnfr
["ip-address"] = vnfr_update
["ip-address"] = vnf_RO
["ip_address"].split(";")[0]
653 elif not db_vnfr
.get("ip-address"):
654 if db_vnfr
.get("vdur"): # if not VDUs, there is not ip_address
655 raise LcmExceptionNoMgmtIP("ns member_vnf_index '{}' has no IP address".format(vnf_index
))
657 for vdu_index
, vdur
in enumerate(get_iterable(db_vnfr
, "vdur")):
658 vdur_RO_count_index
= 0
659 if vdur
.get("pdu-type"):
661 for vdur_RO
in get_iterable(vnf_RO
, "vms"):
662 if vdur
["vdu-id-ref"] != vdur_RO
["vdu_osm_id"]:
664 if vdur
["count-index"] != vdur_RO_count_index
:
665 vdur_RO_count_index
+= 1
667 vdur
["vim-id"] = vdur_RO
.get("vim_vm_id")
668 if vdur_RO
.get("ip_address"):
669 vdur
["ip-address"] = vdur_RO
["ip_address"].split(";")[0]
671 vdur
["ip-address"] = None
672 vdur
["vdu-id-ref"] = vdur_RO
.get("vdu_osm_id")
673 vdur
["name"] = vdur_RO
.get("vim_name")
674 vdur
["status"] = vdur_RO
.get("status")
675 vdur
["status-detailed"] = vdur_RO
.get("error_msg")
676 for ifacer
in get_iterable(vdur
, "interfaces"):
677 for interface_RO
in get_iterable(vdur_RO
, "interfaces"):
678 if ifacer
["name"] == interface_RO
.get("internal_name"):
679 ifacer
["ip-address"] = interface_RO
.get("ip_address")
680 ifacer
["mac-address"] = interface_RO
.get("mac_address")
683 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} interface={} "
685 .format(vnf_index
, vdur
["vdu-id-ref"], ifacer
["name"]))
686 vnfr_update
["vdur.{}".format(vdu_index
)] = vdur
689 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vdur={} count_index={} from "
690 "VIM info".format(vnf_index
, vdur
["vdu-id-ref"], vdur
["count-index"]))
692 for vld_index
, vld
in enumerate(get_iterable(db_vnfr
, "vld")):
693 for net_RO
in get_iterable(nsr_desc_RO
, "nets"):
694 if vld
["id"] != net_RO
.get("vnf_net_osm_id"):
696 vld
["vim-id"] = net_RO
.get("vim_net_id")
697 vld
["name"] = net_RO
.get("vim_name")
698 vld
["status"] = net_RO
.get("status")
699 vld
["status-detailed"] = net_RO
.get("error_msg")
700 vnfr_update
["vld.{}".format(vld_index
)] = vld
703 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} vld={} from VIM info".format(
704 vnf_index
, vld
["id"]))
706 self
.update_db_2("vnfrs", db_vnfr
["_id"], vnfr_update
)
710 raise LcmException("ns_update_vnfr: Not found member_vnf_index={} from VIM info".format(vnf_index
))
712 def _get_ns_config_info(self
, nsr_id
):
714 Generates a mapping between vnf,vdu elements and the N2VC id
715 :param nsr_id: id of nsr to get last database _admin.deployed.VCA that contains this list
716 :return: a dictionary with {osm-config-mapping: {}} where its element contains:
717 "<member-vnf-index>": <N2VC-id> for a vnf configuration, or
718 "<member-vnf-index>.<vdu.id>.<vdu replica(0, 1,..)>": <N2VC-id> for a vdu configuration
720 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
721 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
723 ns_config_info
= {"osm-config-mapping": mapping
}
724 for vca
in vca_deployed_list
:
725 if not vca
["member-vnf-index"]:
727 if not vca
["vdu_id"]:
728 mapping
[vca
["member-vnf-index"]] = vca
["application"]
730 mapping
["{}.{}.{}".format(vca
["member-vnf-index"], vca
["vdu_id"], vca
["vdu_count_index"])] =\
732 return ns_config_info
735 def _get_initial_config_primitive_list(desc_primitive_list
, vca_deployed
):
737 Generates a list of initial-config-primitive based on the list provided by the descriptor. It includes internal
738 primitives as verify-ssh-credentials, or config when needed
739 :param desc_primitive_list: information of the descriptor
740 :param vca_deployed: information of the deployed, needed for known if it is related to an NS, VNF, VDU and if
741 this element contains a ssh public key
742 :return: The modified list. Can ba an empty list, but always a list
744 if desc_primitive_list
:
745 primitive_list
= desc_primitive_list
.copy()
748 # look for primitive config, and get the position. None if not present
749 config_position
= None
750 for index
, primitive
in enumerate(primitive_list
):
751 if primitive
["name"] == "config":
752 config_position
= index
755 # for NS, add always a config primitive if not present (bug 874)
756 if not vca_deployed
["member-vnf-index"] and config_position
is None:
757 primitive_list
.insert(0, {"name": "config", "parameter": []})
759 # for VNF/VDU add verify-ssh-credentials after config
760 if vca_deployed
["member-vnf-index"] and config_position
is not None and vca_deployed
.get("ssh-public-key"):
761 primitive_list
.insert(config_position
+ 1, {"name": "verify-ssh-credentials", "parameter": []})
762 return primitive_list
764 async def _instantiate_ng_ro(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
765 n2vc_key_list
, stage
, start_deploy
, timeout_ns_deploy
):
766 nslcmop_id
= db_nslcmop
["_id"]
768 "name": db_nsr
["name"],
771 "image": deepcopy(db_nsr
["image"]),
772 "flavor": deepcopy(db_nsr
["flavor"]),
773 "action_id": nslcmop_id
,
775 for image
in target
["image"]:
776 image
["vim_info"] = []
777 for flavor
in target
["flavor"]:
778 flavor
["vim_info"] = []
780 ns_params
= db_nslcmop
.get("operationParams")
782 if ns_params
.get("ssh_keys"):
783 ssh_keys
+= ns_params
.get("ssh_keys")
785 ssh_keys
+= n2vc_key_list
788 for vld_index
, vld
in enumerate(nsd
.get("vld")):
789 target_vld
= {"id": vld
["id"],
791 "mgmt-network": vld
.get("mgmt-network", False),
792 "type": vld
.get("type"),
793 "vim_info": [{"vim-network-name": vld
.get("vim-network-name"),
794 "vim_account_id": ns_params
["vimAccountId"]}],
796 for cp
in vld
["vnfd-connection-point-ref"]:
797 cp2target
["member_vnf:{}.{}".format(cp
["member-vnf-index-ref"], cp
["vnfd-connection-point-ref"])] = \
798 "nsrs:{}:vld.{}".format(nsr_id
, vld_index
)
799 target
["ns"]["vld"].append(target_vld
)
800 for vnfr
in db_vnfrs
.values():
801 vnfd
= db_vnfds_ref
[vnfr
["vnfd-ref"]]
802 target_vnf
= deepcopy(vnfr
)
803 for vld
in target_vnf
.get("vld", ()):
804 # check if connected to a ns.vld
805 vnf_cp
= next((cp
for cp
in vnfd
.get("connection-point", ()) if
806 cp
.get("internal-vld-ref") == vld
["id"]), None)
808 ns_cp
= "member_vnf:{}.{}".format(vnfr
["member-vnf-index-ref"], vnf_cp
["id"])
809 if cp2target
.get(ns_cp
):
810 vld
["target"] = cp2target
[ns_cp
]
811 vld
["vim_info"] = [{"vim-network-name": vld
.get("vim-network-name"),
812 "vim_account_id": vnfr
["vim-account-id"]}]
814 for vdur
in target_vnf
.get("vdur", ()):
815 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
816 vdud_index
, vdud
= next(k
for k
in enumerate(vnfd
["vdu"]) if k
[1]["id"] == vdur
["vdu-id-ref"])
817 # vdur["additionalParams"] = vnfr.get("additionalParamsForVnf") # TODO additional params for VDU
820 if deep_get(vdud
, ("vdu-configuration", "config-access", "ssh-access", "required")):
821 vdur
["ssh-keys"] = ssh_keys
822 vdur
["ssh-access-required"] = True
823 elif deep_get(vnfd
, ("vnf-configuration", "config-access", "ssh-access", "required")) and \
824 any(iface
.get("mgmt-vnf") for iface
in vdur
["interfaces"]):
825 vdur
["ssh-keys"] = ssh_keys
826 vdur
["ssh-access-required"] = True
829 if vdud
.get("cloud-init-file"):
830 vdur
["cloud-init"] = "{}:file:{}".format(vnfd
["_id"], vdud
.get("cloud-init-file"))
831 elif vdud
.get("cloud-init"):
832 vdur
["cloud-init"] = "{}:vdu:{}".format(vnfd
["_id"], vdud_index
)
835 ns_flavor
= target
["flavor"][int(vdur
["ns-flavor-id"])]
836 if not next((vi
for vi
in ns_flavor
["vim_info"] if
837 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
838 ns_flavor
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
840 ns_image
= target
["image"][int(vdur
["ns-image-id"])]
841 if not next((vi
for vi
in ns_image
["vim_info"] if
842 vi
and vi
.get("vim_account_id") == vnfr
["vim-account-id"]), None):
843 ns_image
["vim_info"].append({"vim_account_id": vnfr
["vim-account-id"]})
845 vdur
["vim_info"] = [{"vim_account_id": vnfr
["vim-account-id"]}]
846 target
["vnf"].append(target_vnf
)
848 desc
= await self
.RO
.deploy(nsr_id
, target
)
849 action_id
= desc
["action_id"]
850 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, timeout_ns_deploy
, stage
)
854 "_admin.deployed.RO.operational-status": "running",
855 "detailed-status": " ".join(stage
)
857 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
858 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
859 self
._write
_op
_status
(nslcmop_id
, stage
)
860 self
.logger
.debug(logging_text
+ "ns deployed at RO. RO_id={}".format(action_id
))
863 async def _wait_ng_ro(self
, nsr_id
, action_id
, nslcmop_id
, start_time
, timeout
, stage
):
864 detailed_status_old
= None
866 while time() <= start_time
+ timeout
:
867 desc_status
= await self
.RO
.status(nsr_id
, action_id
)
868 if desc_status
["status"] == "FAILED":
869 raise NgRoException(desc_status
["details"])
870 elif desc_status
["status"] == "BUILD":
871 stage
[2] = "VIM: ({})".format(desc_status
["details"])
872 elif desc_status
["status"] == "DONE":
873 stage
[2] = "Deployed at VIM"
876 assert False, "ROclient.check_ns_status returns unknown {}".format(desc_status
["status"])
877 if stage
[2] != detailed_status_old
:
878 detailed_status_old
= stage
[2]
879 db_nsr_update
["detailed-status"] = " ".join(stage
)
880 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
881 self
._write
_op
_status
(nslcmop_id
, stage
)
882 await asyncio
.sleep(5, loop
=self
.loop
)
883 else: # timeout_ns_deploy
884 raise NgRoException("Timeout waiting ns to deploy")
886 async def _terminate_ng_ro(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
890 start_deploy
= time()
898 desc
= await self
.RO
.deploy(nsr_id
, target
)
899 action_id
= desc
["action_id"]
900 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = action_id
901 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETING"
902 self
.logger
.debug(logging_text
+ "ns terminate action at RO. action_id={}".format(action_id
))
905 delete_timeout
= 20 * 60 # 20 minutes
906 await self
._wait
_ng
_ro
(self
, nsr_id
, action_id
, nslcmop_id
, start_deploy
, delete_timeout
, stage
)
908 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
909 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
911 await self
.RO
.delete(nsr_id
)
912 except Exception as e
:
913 if isinstance(e
, NgRoException
) and e
.http_code
== 404: # not found
914 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
915 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
916 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
917 self
.logger
.debug(logging_text
+ "RO_action_id={} already deleted".format(action_id
))
918 elif isinstance(e
, NgRoException
) and e
.http_code
== 409: # conflict
919 failed_detail
.append("delete conflict: {}".format(e
))
920 self
.logger
.debug(logging_text
+ "RO_action_id={} delete conflict: {}".format(action_id
, e
))
922 failed_detail
.append("delete error: {}".format(e
))
923 self
.logger
.error(logging_text
+ "RO_action_id={} delete error: {}".format(action_id
, e
))
926 stage
[2] = "Error deleting from VIM"
928 stage
[2] = "Deleted from VIM"
929 db_nsr_update
["detailed-status"] = " ".join(stage
)
930 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
931 self
._write
_op
_status
(nslcmop_id
, stage
)
934 raise LcmException("; ".join(failed_detail
))
937 async def instantiate_RO(self
, logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
, db_vnfds_ref
,
938 n2vc_key_list
, stage
):
941 :param logging_text: preffix text to use at logging
942 :param nsr_id: nsr identity
943 :param nsd: database content of ns descriptor
944 :param db_nsr: database content of ns record
945 :param db_nslcmop: database content of ns operation, in this case, 'instantiate'
947 :param db_vnfds_ref: database content of vnfds, indexed by id (not _id). {id: {vnfd_object}, ...}
948 :param n2vc_key_list: ssh-public-key list to be inserted to management vdus via cloud-init
949 :param stage: list with 3 items: [general stage, tasks, vim_specific]. This task will write over vim_specific
950 :return: None or exception
954 RO_descriptor_number
= 0 # number of descriptors created at RO
955 vnf_index_2_RO_id
= {} # map between vnfd/nsd id to the id used at RO
956 nslcmop_id
= db_nslcmop
["_id"]
957 start_deploy
= time()
958 ns_params
= db_nslcmop
.get("operationParams")
959 if ns_params
and ns_params
.get("timeout_ns_deploy"):
960 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
962 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
964 # Check for and optionally request placement optimization. Database will be updated if placement activated
965 stage
[2] = "Waiting for Placement."
966 if await self
._do
_placement
(logging_text
, db_nslcmop
, db_vnfrs
):
967 # in case of placement change ns_params[vimAcountId) if not present at any vnfrs
968 for vnfr
in db_vnfrs
.values():
969 if ns_params
["vimAccountId"] == vnfr
["vim-account-id"]:
972 ns_params
["vimAccountId"] == vnfr
["vim-account-id"]
975 return await self
._instantiate
_ng
_ro
(logging_text
, nsr_id
, nsd
, db_nsr
, db_nslcmop
, db_vnfrs
,
976 db_vnfds_ref
, n2vc_key_list
, stage
, start_deploy
,
979 # get vnfds, instantiate at RO
980 for c_vnf
in nsd
.get("constituent-vnfd", ()):
981 member_vnf_index
= c_vnf
["member-vnf-index"]
982 vnfd
= db_vnfds_ref
[c_vnf
['vnfd-id-ref']]
983 vnfd_ref
= vnfd
["id"]
985 stage
[2] = "Creating vnfd='{}' member_vnf_index='{}' at RO".format(vnfd_ref
, member_vnf_index
)
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
)
990 # self.logger.debug(logging_text + stage[2])
991 vnfd_id_RO
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, member_vnf_index
[:23])
992 vnf_index_2_RO_id
[member_vnf_index
] = vnfd_id_RO
993 RO_descriptor_number
+= 1
995 # look position at deployed.RO.vnfd if not present it will be appended at the end
996 for index
, vnf_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"]):
997 if vnf_deployed
["member-vnf-index"] == member_vnf_index
:
1000 index
= len(db_nsr
["_admin"]["deployed"]["RO"]["vnfd"])
1001 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"].append(None)
1004 RO_update
= {"member-vnf-index": member_vnf_index
}
1005 vnfd_list
= await self
.RO
.get_list("vnfd", filter_by
={"osm_id": vnfd_id_RO
})
1007 RO_update
["id"] = vnfd_list
[0]["uuid"]
1008 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' exists at RO. Using RO_id={}".
1009 format(vnfd_ref
, member_vnf_index
, vnfd_list
[0]["uuid"]))
1011 vnfd_RO
= self
.vnfd2RO(vnfd
, vnfd_id_RO
, db_vnfrs
[c_vnf
["member-vnf-index"]].
1012 get("additionalParamsForVnf"), nsr_id
)
1013 desc
= await self
.RO
.create("vnfd", descriptor
=vnfd_RO
)
1014 RO_update
["id"] = desc
["uuid"]
1015 self
.logger
.debug(logging_text
+ "vnfd='{}' member_vnf_index='{}' created at RO. RO_id={}".format(
1016 vnfd_ref
, member_vnf_index
, desc
["uuid"]))
1017 db_nsr_update
["_admin.deployed.RO.vnfd.{}".format(index
)] = RO_update
1018 db_nsr
["_admin"]["deployed"]["RO"]["vnfd"][index
] = RO_update
1023 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1024 db_nsr_update
["detailed-status"] = " ".join(stage
)
1025 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1026 self
._write
_op
_status
(nslcmop_id
, stage
)
1028 # self.logger.debug(logging_text + stage[2])
1029 RO_osm_nsd_id
= "{}.{}.{}".format(nsr_id
, RO_descriptor_number
, nsd_ref
[:23])
1030 RO_descriptor_number
+= 1
1031 nsd_list
= await self
.RO
.get_list("nsd", filter_by
={"osm_id": RO_osm_nsd_id
})
1033 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= nsd_list
[0]["uuid"]
1034 self
.logger
.debug(logging_text
+ "nsd={} exists at RO. Using RO_id={}".format(
1035 nsd_ref
, RO_nsd_uuid
))
1037 nsd_RO
= deepcopy(nsd
)
1038 nsd_RO
["id"] = RO_osm_nsd_id
1039 nsd_RO
.pop("_id", None)
1040 nsd_RO
.pop("_admin", None)
1041 for c_vnf
in nsd_RO
.get("constituent-vnfd", ()):
1042 member_vnf_index
= c_vnf
["member-vnf-index"]
1043 c_vnf
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1044 for c_vld
in nsd_RO
.get("vld", ()):
1045 for cp
in c_vld
.get("vnfd-connection-point-ref", ()):
1046 member_vnf_index
= cp
["member-vnf-index-ref"]
1047 cp
["vnfd-id-ref"] = vnf_index_2_RO_id
[member_vnf_index
]
1049 desc
= await self
.RO
.create("nsd", descriptor
=nsd_RO
)
1050 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1051 db_nsr_update
["_admin.deployed.RO.nsd_id"] = RO_nsd_uuid
= desc
["uuid"]
1052 self
.logger
.debug(logging_text
+ "nsd={} created at RO. RO_id={}".format(nsd_ref
, RO_nsd_uuid
))
1053 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1056 stage
[2] = "Creating nsd={} at RO".format(nsd_ref
)
1057 db_nsr_update
["detailed-status"] = " ".join(stage
)
1058 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1059 self
._write
_op
_status
(nslcmop_id
, stage
)
1061 # if present use it unless in error status
1062 RO_nsr_id
= deep_get(db_nsr
, ("_admin", "deployed", "RO", "nsr_id"))
1065 stage
[2] = "Looking for existing ns at RO"
1066 db_nsr_update
["detailed-status"] = " ".join(stage
)
1067 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1068 self
._write
_op
_status
(nslcmop_id
, stage
)
1069 # self.logger.debug(logging_text + stage[2] + " RO_ns_id={}".format(RO_nsr_id))
1070 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1072 except ROclient
.ROClientException
as e
:
1073 if e
.http_code
!= HTTPStatus
.NOT_FOUND
:
1075 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1077 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1078 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1079 if ns_status
== "ERROR":
1080 stage
[2] = "Deleting ns at RO. RO_ns_id={}".format(RO_nsr_id
)
1081 self
.logger
.debug(logging_text
+ stage
[2])
1082 await self
.RO
.delete("ns", RO_nsr_id
)
1083 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
1085 stage
[2] = "Checking dependencies"
1086 db_nsr_update
["detailed-status"] = " ".join(stage
)
1087 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1088 self
._write
_op
_status
(nslcmop_id
, stage
)
1089 # self.logger.debug(logging_text + stage[2])
1091 # check if VIM is creating and wait look if previous tasks in process
1092 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account", ns_params
["vimAccountId"])
1094 stage
[2] = "Waiting for related tasks '{}' to be completed".format(task_name
)
1095 self
.logger
.debug(logging_text
+ stage
[2])
1096 await asyncio
.wait(task_dependency
, timeout
=3600)
1097 if ns_params
.get("vnf"):
1098 for vnf
in ns_params
["vnf"]:
1099 if "vimAccountId" in vnf
:
1100 task_name
, task_dependency
= self
.lcm_tasks
.lookfor_related("vim_account",
1101 vnf
["vimAccountId"])
1103 stage
[2] = "Waiting for related tasks '{}' to be completed.".format(task_name
)
1104 self
.logger
.debug(logging_text
+ stage
[2])
1105 await asyncio
.wait(task_dependency
, timeout
=3600)
1107 stage
[2] = "Checking instantiation parameters."
1108 RO_ns_params
= self
._ns
_params
_2_RO
(ns_params
, nsd
, db_vnfds_ref
, db_vnfrs
, n2vc_key_list
)
1109 stage
[2] = "Deploying ns at VIM."
1110 db_nsr_update
["detailed-status"] = " ".join(stage
)
1111 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1112 self
._write
_op
_status
(nslcmop_id
, stage
)
1114 desc
= await self
.RO
.create("ns", descriptor
=RO_ns_params
, name
=db_nsr
["name"], scenario
=RO_nsd_uuid
)
1115 RO_nsr_id
= db_nsr_update
["_admin.deployed.RO.nsr_id"] = desc
["uuid"]
1116 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1117 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "BUILD"
1118 self
.logger
.debug(logging_text
+ "ns created at RO. RO_id={}".format(desc
["uuid"]))
1120 # wait until NS is ready
1121 stage
[2] = "Waiting VIM to deploy ns."
1122 db_nsr_update
["detailed-status"] = " ".join(stage
)
1123 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1124 self
._write
_op
_status
(nslcmop_id
, stage
)
1125 detailed_status_old
= None
1126 self
.logger
.debug(logging_text
+ stage
[2] + " RO_ns_id={}".format(RO_nsr_id
))
1129 while time() <= start_deploy
+ timeout_ns_deploy
:
1130 desc
= await self
.RO
.show("ns", RO_nsr_id
)
1133 if desc
!= old_desc
:
1134 # desc has changed => update db
1135 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
1138 ns_status
, ns_status_info
= self
.RO
.check_ns_status(desc
)
1139 db_nsr_update
["_admin.deployed.RO.nsr_status"] = ns_status
1140 if ns_status
== "ERROR":
1141 raise ROclient
.ROClientException(ns_status_info
)
1142 elif ns_status
== "BUILD":
1143 stage
[2] = "VIM: ({})".format(ns_status_info
)
1144 elif ns_status
== "ACTIVE":
1145 stage
[2] = "Waiting for management IP address reported by the VIM. Updating VNFRs."
1147 self
.ns_update_vnfr(db_vnfrs
, desc
)
1149 except LcmExceptionNoMgmtIP
:
1152 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
1153 if stage
[2] != detailed_status_old
:
1154 detailed_status_old
= stage
[2]
1155 db_nsr_update
["detailed-status"] = " ".join(stage
)
1156 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1157 self
._write
_op
_status
(nslcmop_id
, stage
)
1158 await asyncio
.sleep(5, loop
=self
.loop
)
1159 else: # timeout_ns_deploy
1160 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
1163 self
.ns_update_nsr(db_nsr_update
, db_nsr
, desc
)
1165 db_nsr_update
["_admin.deployed.RO.operational-status"] = "running"
1166 # db_nsr["_admin.deployed.RO.detailed-status"] = "Deployed at VIM"
1167 stage
[2] = "Deployed at VIM"
1168 db_nsr_update
["detailed-status"] = " ".join(stage
)
1169 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1170 self
._write
_op
_status
(nslcmop_id
, stage
)
1171 # await self._on_update_n2vc_db("nsrs", {"_id": nsr_id}, "_admin.deployed", db_nsr_update)
1172 # self.logger.debug(logging_text + "Deployed at VIM")
1173 except (ROclient
.ROClientException
, LcmException
, DbException
, NgRoException
) as e
:
1174 stage
[2] = "ERROR deploying at VIM"
1175 self
.set_vnfr_at_error(db_vnfrs
, str(e
))
1178 async def wait_vm_up_insert_key_ro(self
, logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
, pub_key
=None, user
=None):
1180 Wait for ip addres at RO, and optionally, insert public key in virtual machine
1181 :param logging_text: prefix use for logging
1186 :param pub_key: public ssh key to inject, None to skip
1187 :param user: user to apply the public ssh key
1191 # self.logger.debug(logging_text + "Starting wait_vm_up_insert_key_ro")
1195 target_vdu_id
= None
1201 if ro_retries
>= 360: # 1 hour
1202 raise LcmException("Not found _admin.deployed.RO.nsr_id for nsr_id: {}".format(nsr_id
))
1204 await asyncio
.sleep(10, loop
=self
.loop
)
1207 if not target_vdu_id
:
1208 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnfr_id
})
1210 if not vdu_id
: # for the VNF case
1211 if db_vnfr
.get("status") == "ERROR":
1212 raise LcmException("Cannot inject ssh-key because target VNF is in error state")
1213 ip_address
= db_vnfr
.get("ip-address")
1216 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur") if x
.get("ip-address") == ip_address
), None)
1218 vdur
= next((x
for x
in get_iterable(db_vnfr
, "vdur")
1219 if x
.get("vdu-id-ref") == vdu_id
and x
.get("count-index") == vdu_index
), None)
1221 if not vdur
and len(db_vnfr
.get("vdur", ())) == 1: # If only one, this should be the target vdu
1222 vdur
= db_vnfr
["vdur"][0]
1224 raise LcmException("Not found vnfr_id={}, vdu_id={}, vdu_index={}".format(vnfr_id
, vdu_id
,
1227 if vdur
.get("pdu-type") or vdur
.get("status") == "ACTIVE":
1228 ip_address
= vdur
.get("ip-address")
1231 target_vdu_id
= vdur
["vdu-id-ref"]
1232 elif vdur
.get("status") == "ERROR":
1233 raise LcmException("Cannot inject ssh-key because target VM is in error state")
1235 if not target_vdu_id
:
1238 # inject public key into machine
1239 if pub_key
and user
:
1240 # wait until NS is deployed at RO
1242 db_nsrs
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1243 ro_nsr_id
= deep_get(db_nsrs
, ("_admin", "deployed", "RO", "nsr_id"))
1247 # self.logger.debug(logging_text + "Inserting RO key")
1248 if vdur
.get("pdu-type"):
1249 self
.logger
.error(logging_text
+ "Cannot inject ssh-ky to a PDU")
1252 ro_vm_id
= "{}-{}".format(db_vnfr
["member-vnf-index-ref"], target_vdu_id
) # TODO add vdu_index
1254 target
= {"action": "inject_ssh_key", "key": pub_key
, "user": user
,
1255 "vnf": [{"_id": vnfr_id
, "vdur": [{"id": vdu_id
}]}],
1257 await self
.RO
.deploy(nsr_id
, target
)
1259 result_dict
= await self
.RO
.create_action(
1261 item_id_name
=ro_nsr_id
,
1262 descriptor
={"add_public_key": pub_key
, "vms": [ro_vm_id
], "user": user
}
1264 # result_dict contains the format {VM-id: {vim_result: 200, description: text}}
1265 if not result_dict
or not isinstance(result_dict
, dict):
1266 raise LcmException("Unknown response from RO when injecting key")
1267 for result
in result_dict
.values():
1268 if result
.get("vim_result") == 200:
1271 raise ROclient
.ROClientException("error injecting key: {}".format(
1272 result
.get("description")))
1274 except NgRoException
as e
:
1275 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1276 except ROclient
.ROClientException
as e
:
1278 self
.logger
.debug(logging_text
+ "error injecting key: {}. Retrying until {} seconds".
1282 raise LcmException("Reaching max tries injecting key. Error: {}".format(e
))
1288 async def _wait_dependent_n2vc(self
, nsr_id
, vca_deployed_list
, vca_index
):
1290 Wait until dependent VCA deployments have been finished. NS wait for VNFs and VDUs. VNFs for VDUs
1292 my_vca
= vca_deployed_list
[vca_index
]
1293 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1294 # vdu or kdu: no dependencies
1298 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1299 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1300 configuration_status_list
= db_nsr
["configurationStatus"]
1301 for index
, vca_deployed
in enumerate(configuration_status_list
):
1302 if index
== vca_index
:
1305 if not my_vca
.get("member-vnf-index") or \
1306 (vca_deployed
.get("member-vnf-index") == my_vca
.get("member-vnf-index")):
1307 internal_status
= configuration_status_list
[index
].get("status")
1308 if internal_status
== 'READY':
1310 elif internal_status
== 'BROKEN':
1311 raise LcmException("Configuration aborted because dependent charm/s has failed")
1315 # no dependencies, return
1317 await asyncio
.sleep(10)
1320 raise LcmException("Configuration aborted because dependent charm/s timeout")
1322 async def instantiate_N2VC(self
, logging_text
, vca_index
, nsi_id
, db_nsr
, db_vnfr
, vdu_id
, kdu_name
, vdu_index
,
1323 config_descriptor
, deploy_params
, base_folder
, nslcmop_id
, stage
):
1324 nsr_id
= db_nsr
["_id"]
1325 db_update_entry
= "_admin.deployed.VCA.{}.".format(vca_index
)
1326 vca_deployed_list
= db_nsr
["_admin"]["deployed"]["VCA"]
1327 vca_deployed
= db_nsr
["_admin"]["deployed"]["VCA"][vca_index
]
1329 'collection': 'nsrs',
1330 'filter': {'_id': nsr_id
},
1331 'path': db_update_entry
1337 element_under_configuration
= nsr_id
1341 vnfr_id
= db_vnfr
["_id"]
1343 namespace
= "{nsi}.{ns}".format(
1344 nsi
=nsi_id
if nsi_id
else "",
1348 element_type
= 'VNF'
1349 element_under_configuration
= vnfr_id
1350 namespace
+= ".{}".format(vnfr_id
)
1352 namespace
+= ".{}-{}".format(vdu_id
, vdu_index
or 0)
1353 element_type
= 'VDU'
1354 element_under_configuration
= "{}-{}".format(vdu_id
, vdu_index
or 0)
1356 namespace
+= ".{}".format(kdu_name
)
1357 element_type
= 'KDU'
1358 element_under_configuration
= kdu_name
1361 artifact_path
= "{}/{}/charms/{}".format(
1362 base_folder
["folder"],
1363 base_folder
["pkg-dir"],
1364 config_descriptor
["juju"]["charm"]
1367 is_proxy_charm
= deep_get(config_descriptor
, ('juju', 'charm')) is not None
1368 is_k8s_proxy_charm
= False
1370 if deep_get(config_descriptor
, ('juju', 'proxy')) is False:
1371 is_proxy_charm
= False
1373 if deep_get(config_descriptor
, ('juju', 'k8s')) is True and is_proxy_charm
:
1374 is_k8s_proxy_charm
= True
1376 if not is_k8s_proxy_charm
:
1377 # n2vc_redesign STEP 3.1
1379 # find old ee_id if exists
1380 ee_id
= vca_deployed
.get("ee_id")
1382 # create or register execution environment in VCA
1385 self
._write
_configuration
_status
(
1387 vca_index
=vca_index
,
1389 element_under_configuration
=element_under_configuration
,
1390 element_type
=element_type
1393 step
= "create execution environment"
1394 self
.logger
.debug(logging_text
+ step
)
1395 ee_id
, credentials
= await self
.n2vc
.create_execution_environment(namespace
=namespace
,
1400 step
= "Waiting to VM being up and getting IP address"
1401 self
.logger
.debug(logging_text
+ step
)
1402 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1403 user
=None, pub_key
=None)
1404 credentials
= {"hostname": rw_mgmt_ip
}
1406 username
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1407 # TODO remove this when changes on IM regarding config-access:ssh-access:default-user were
1408 # merged. Meanwhile let's get username from initial-config-primitive
1409 if not username
and config_descriptor
.get("initial-config-primitive"):
1410 for config_primitive
in config_descriptor
["initial-config-primitive"]:
1411 for param
in config_primitive
.get("parameter", ()):
1412 if param
["name"] == "ssh-username":
1413 username
= param
["value"]
1416 raise LcmException("Cannot determine the username neither with"
1417 "'initial-config-promitive' nor with "
1418 "'config-access.ssh-access.default-user'")
1419 credentials
["username"] = username
1420 # n2vc_redesign STEP 3.2
1422 self
._write
_configuration
_status
(
1424 vca_index
=vca_index
,
1425 status
='REGISTERING',
1426 element_under_configuration
=element_under_configuration
,
1427 element_type
=element_type
1430 step
= "register execution environment {}".format(credentials
)
1431 self
.logger
.debug(logging_text
+ step
)
1432 ee_id
= await self
.n2vc
.register_execution_environment(credentials
=credentials
, namespace
=namespace
,
1435 # n2vc_redesign STEP 3.3
1437 step
= "Install configuration Software"
1439 self
._write
_configuration
_status
(
1441 vca_index
=vca_index
,
1442 status
='INSTALLING SW',
1443 element_under_configuration
=element_under_configuration
,
1444 element_type
=element_type
,
1447 # TODO check if already done
1448 self
.logger
.debug(logging_text
+ step
)
1450 if not is_proxy_charm
:
1451 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1452 if initial_config_primitive_list
:
1453 for primitive
in initial_config_primitive_list
:
1454 if primitive
["name"] == "config":
1455 config
= self
._map
_primitive
_params
(
1461 if is_k8s_proxy_charm
:
1462 charm_name
= deep_get(config_descriptor
, ('juju', 'charm'))
1463 self
.logger
.debug("Installing K8s Proxy Charm: {}".format(charm_name
))
1465 ee_id
= await self
.n2vc
.install_k8s_proxy_charm(
1466 charm_name
=charm_name
,
1467 namespace
=namespace
,
1468 artifact_path
=artifact_path
,
1474 if element_type
== "NS":
1475 num_units
= db_nsr
.get("config-units") or 1
1476 elif element_type
== "VNF":
1477 num_units
= db_vnfr
.get("config-units") or 1
1478 elif element_type
== "VDU":
1479 for v
in db_vnfr
["vdur"]:
1480 if vdu_id
== v
["vdu-id-ref"]:
1481 num_units
= v
.get("config-units") or 1
1484 await self
.n2vc
.install_configuration_sw(
1486 artifact_path
=artifact_path
,
1492 # write in db flag of configuration_sw already installed
1493 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "config_sw_installed": True})
1495 # add relations for this VCA (wait for other peers related with this VCA)
1496 await self
._add
_vca
_relations
(logging_text
=logging_text
, nsr_id
=nsr_id
, vca_index
=vca_index
)
1498 # if SSH access is required, then get execution environment SSH public
1499 if is_proxy_charm
: # if native charm we have waited already to VM be UP
1502 if deep_get(config_descriptor
, ("config-access", "ssh-access", "required")):
1503 # Needed to inject a ssh key
1504 user
= deep_get(config_descriptor
, ("config-access", "ssh-access", "default-user"))
1505 step
= "Install configuration Software, getting public ssh key"
1506 pub_key
= await self
.n2vc
.get_ee_ssh_public__key(ee_id
=ee_id
, db_dict
=db_dict
)
1508 step
= "Insert public key into VM user={} ssh_key={}".format(user
, pub_key
)
1510 step
= "Waiting to VM being up and getting IP address"
1511 self
.logger
.debug(logging_text
+ step
)
1513 # n2vc_redesign STEP 5.1
1514 # wait for RO (ip-address) Insert pub_key into VM
1516 rw_mgmt_ip
= await self
.wait_vm_up_insert_key_ro(logging_text
, nsr_id
, vnfr_id
, vdu_id
, vdu_index
,
1517 user
=user
, pub_key
=pub_key
)
1519 rw_mgmt_ip
= None # This is for a NS configuration
1521 self
.logger
.debug(logging_text
+ ' VM_ip_address={}'.format(rw_mgmt_ip
))
1523 # store rw_mgmt_ip in deploy params for later replacement
1524 deploy_params
["rw_mgmt_ip"] = rw_mgmt_ip
1526 # n2vc_redesign STEP 6 Execute initial config primitive
1527 step
= 'execute initial config primitive'
1528 initial_config_primitive_list
= config_descriptor
.get('initial-config-primitive')
1530 # sort initial config primitives by 'seq'
1531 if initial_config_primitive_list
:
1533 initial_config_primitive_list
.sort(key
=lambda val
: int(val
['seq']))
1534 except Exception as e
:
1535 self
.logger
.error(logging_text
+ step
+ ": " + str(e
))
1537 self
.logger
.debug(logging_text
+ step
+ ": No initial-config-primitive")
1539 # add config if not present for NS charm
1540 initial_config_primitive_list
= self
._get
_initial
_config
_primitive
_list
(initial_config_primitive_list
,
1543 # wait for dependent primitives execution (NS -> VNF -> VDU)
1544 if initial_config_primitive_list
:
1545 await self
._wait
_dependent
_n
2vc
(nsr_id
, vca_deployed_list
, vca_index
)
1547 # stage, in function of element type: vdu, kdu, vnf or ns
1548 my_vca
= vca_deployed_list
[vca_index
]
1549 if my_vca
.get("vdu_id") or my_vca
.get("kdu_name"):
1551 stage
[0] = 'Stage 3/5: running Day-1 primitives for VDU.'
1552 elif my_vca
.get("member-vnf-index"):
1554 stage
[0] = 'Stage 4/5: running Day-1 primitives for VNF.'
1557 stage
[0] = 'Stage 5/5: running Day-1 primitives for NS.'
1559 self
._write
_configuration
_status
(
1561 vca_index
=vca_index
,
1562 status
='EXECUTING PRIMITIVE'
1565 self
._write
_op
_status
(
1570 check_if_terminated_needed
= True
1571 for initial_config_primitive
in initial_config_primitive_list
:
1572 # adding information on the vca_deployed if it is a NS execution environment
1573 if not vca_deployed
["member-vnf-index"]:
1574 deploy_params
["ns_config_info"] = json
.dumps(self
._get
_ns
_config
_info
(nsr_id
))
1575 # TODO check if already done
1576 primitive_params_
= self
._map
_primitive
_params
(initial_config_primitive
, {}, deploy_params
)
1578 step
= "execute primitive '{}' params '{}'".format(initial_config_primitive
["name"], primitive_params_
)
1579 self
.logger
.debug(logging_text
+ step
)
1580 await self
.n2vc
.exec_primitive(
1582 primitive_name
=initial_config_primitive
["name"],
1583 params_dict
=primitive_params_
,
1586 # Once some primitive has been exec, check and write at db if it needs to exec terminated primitives
1587 if check_if_terminated_needed
:
1588 if config_descriptor
.get('terminate-config-primitive'):
1589 self
.update_db_2("nsrs", nsr_id
, {db_update_entry
+ "needed_terminate": True})
1590 check_if_terminated_needed
= False
1592 # TODO register in database that primitive is done
1594 step
= "instantiated at VCA"
1595 self
.logger
.debug(logging_text
+ step
)
1597 self
._write
_configuration
_status
(
1599 vca_index
=vca_index
,
1603 except Exception as e
: # TODO not use Exception but N2VC exception
1604 # self.update_db_2("nsrs", nsr_id, {db_update_entry + "instantiation": "FAILED"})
1605 if not isinstance(e
, (DbException
, N2VCException
, LcmException
, asyncio
.CancelledError
)):
1606 self
.logger
.error("Exception while {} : {}".format(step
, e
), exc_info
=True)
1607 self
._write
_configuration
_status
(
1609 vca_index
=vca_index
,
1612 raise LcmException("{} {}".format(step
, e
)) from e
1614 def _write_ns_status(self
, nsr_id
: str, ns_state
: str, current_operation
: str, current_operation_id
: str,
1615 error_description
: str = None, error_detail
: str = None, other_update
: dict = None):
1617 Update db_nsr fields.
1620 :param current_operation:
1621 :param current_operation_id:
1622 :param error_description:
1623 :param error_detail:
1624 :param other_update: Other required changes at database if provided, will be cleared
1628 db_dict
= other_update
or {}
1629 db_dict
["_admin.nslcmop"] = current_operation_id
# for backward compatibility
1630 db_dict
["_admin.current-operation"] = current_operation_id
1631 db_dict
["_admin.operation-type"] = current_operation
if current_operation
!= "IDLE" else None
1632 db_dict
["currentOperation"] = current_operation
1633 db_dict
["currentOperationID"] = current_operation_id
1634 db_dict
["errorDescription"] = error_description
1635 db_dict
["errorDetail"] = error_detail
1638 db_dict
["nsState"] = ns_state
1639 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1640 except DbException
as e
:
1641 self
.logger
.warn('Error writing NS status, ns={}: {}'.format(nsr_id
, e
))
1643 def _write_op_status(self
, op_id
: str, stage
: list = None, error_message
: str = None, queuePosition
: int = 0,
1644 operation_state
: str = None, other_update
: dict = None):
1646 db_dict
= other_update
or {}
1647 db_dict
['queuePosition'] = queuePosition
1648 if isinstance(stage
, list):
1649 db_dict
['stage'] = stage
[0]
1650 db_dict
['detailed-status'] = " ".join(stage
)
1651 elif stage
is not None:
1652 db_dict
['stage'] = str(stage
)
1654 if error_message
is not None:
1655 db_dict
['errorMessage'] = error_message
1656 if operation_state
is not None:
1657 db_dict
['operationState'] = operation_state
1658 db_dict
["statusEnteredTime"] = time()
1659 self
.update_db_2("nslcmops", op_id
, db_dict
)
1660 except DbException
as e
:
1661 self
.logger
.warn('Error writing OPERATION status for op_id: {} -> {}'.format(op_id
, e
))
1663 def _write_all_config_status(self
, db_nsr
: dict, status
: str):
1665 nsr_id
= db_nsr
["_id"]
1666 # configurationStatus
1667 config_status
= db_nsr
.get('configurationStatus')
1669 db_nsr_update
= {"configurationStatus.{}.status".format(index
): status
for index
, v
in
1670 enumerate(config_status
) if v
}
1672 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1674 except DbException
as e
:
1675 self
.logger
.warn('Error writing all configuration status, ns={}: {}'.format(nsr_id
, e
))
1677 def _write_configuration_status(self
, nsr_id
: str, vca_index
: int, status
: str = None,
1678 element_under_configuration
: str = None, element_type
: str = None,
1679 other_update
: dict = None):
1681 # self.logger.debug('_write_configuration_status(): vca_index={}, status={}'
1682 # .format(vca_index, status))
1685 db_path
= 'configurationStatus.{}.'.format(vca_index
)
1686 db_dict
= other_update
or {}
1688 db_dict
[db_path
+ 'status'] = status
1689 if element_under_configuration
:
1690 db_dict
[db_path
+ 'elementUnderConfiguration'] = element_under_configuration
1692 db_dict
[db_path
+ 'elementType'] = element_type
1693 self
.update_db_2("nsrs", nsr_id
, db_dict
)
1694 except DbException
as e
:
1695 self
.logger
.warn('Error writing configuration status={}, ns={}, vca_index={}: {}'
1696 .format(status
, nsr_id
, vca_index
, e
))
1698 async def _do_placement(self
, logging_text
, db_nslcmop
, db_vnfrs
):
1700 Check and computes the placement, (vim account where to deploy). If it is decided by an external tool, it
1701 sends the request via kafka and wait until the result is wrote at database (nslcmops _admin.plca).
1702 Database is used because the result can be obtained from a different LCM worker in case of HA.
1703 :param logging_text: contains the prefix for logging, with the ns and nslcmop identifiers
1704 :param db_nslcmop: database content of nslcmop
1705 :param db_vnfrs: database content of vnfrs, indexed by member-vnf-index.
1706 :return: True if some modification is done. Modifies database vnfrs and parameter db_vnfr with the
1707 computed 'vim-account-id'
1710 nslcmop_id
= db_nslcmop
['_id']
1711 placement_engine
= deep_get(db_nslcmop
, ('operationParams', 'placement-engine'))
1712 if placement_engine
== "PLA":
1713 self
.logger
.debug(logging_text
+ "Invoke and wait for placement optimization")
1714 await self
.msg
.aiowrite("pla", "get_placement", {'nslcmopId': nslcmop_id
}, loop
=self
.loop
)
1715 db_poll_interval
= 5
1716 wait
= db_poll_interval
* 10
1718 while not pla_result
and wait
>= 0:
1719 await asyncio
.sleep(db_poll_interval
)
1720 wait
-= db_poll_interval
1721 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1722 pla_result
= deep_get(db_nslcmop
, ('_admin', 'pla'))
1725 raise LcmException("Placement timeout for nslcmopId={}".format(nslcmop_id
))
1727 for pla_vnf
in pla_result
['vnf']:
1728 vnfr
= db_vnfrs
.get(pla_vnf
['member-vnf-index'])
1729 if not pla_vnf
.get('vimAccountId') or not vnfr
:
1732 self
.db
.set_one("vnfrs", {"_id": vnfr
["_id"]}, {"vim-account-id": pla_vnf
['vimAccountId']})
1734 vnfr
["vim-account-id"] = pla_vnf
['vimAccountId']
1737 def update_nsrs_with_pla_result(self
, params
):
1739 nslcmop_id
= deep_get(params
, ('placement', 'nslcmopId'))
1740 self
.update_db_2("nslcmops", nslcmop_id
, {"_admin.pla": params
.get('placement')})
1741 except Exception as e
:
1742 self
.logger
.warn('Update failed for nslcmop_id={}:{}'.format(nslcmop_id
, e
))
1744 async def instantiate(self
, nsr_id
, nslcmop_id
):
1747 :param nsr_id: ns instance to deploy
1748 :param nslcmop_id: operation to run
1752 # Try to lock HA task here
1753 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
1754 if not task_is_locked_by_me
:
1755 self
.logger
.debug('instantiate() task is not locked by me, ns={}'.format(nsr_id
))
1758 logging_text
= "Task ns={} instantiate={} ".format(nsr_id
, nslcmop_id
)
1759 self
.logger
.debug(logging_text
+ "Enter")
1764 # get all needed from database
1766 # database nsrs record
1769 # database nslcmops record
1772 # update operation on nsrs
1774 # update operation on nslcmops
1775 db_nslcmop_update
= {}
1777 nslcmop_operation_state
= None
1778 db_vnfrs
= {} # vnf's info indexed by member-index
1780 tasks_dict_info
= {} # from task to info text
1783 stage
= ['Stage 1/5: preparation of the environment.', "Waiting for previous operations to terminate.", ""]
1784 # ^ stage, step, VIM progress
1786 # wait for any previous tasks in process
1787 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
1789 # STEP 0: Reading database (nslcmops, nsrs, nsds, vnfrs, vnfds)
1790 stage
[1] = "Reading from database,"
1791 # nsState="BUILDING", currentOperation="INSTANTIATING", currentOperationID=nslcmop_id
1792 db_nsr_update
["detailed-status"] = "creating"
1793 db_nsr_update
["operational-status"] = "init"
1794 self
._write
_ns
_status
(
1796 ns_state
="BUILDING",
1797 current_operation
="INSTANTIATING",
1798 current_operation_id
=nslcmop_id
,
1799 other_update
=db_nsr_update
1801 self
._write
_op
_status
(
1807 # read from db: operation
1808 stage
[1] = "Getting nslcmop={} from db".format(nslcmop_id
)
1809 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
1810 ns_params
= db_nslcmop
.get("operationParams")
1811 if ns_params
and ns_params
.get("timeout_ns_deploy"):
1812 timeout_ns_deploy
= ns_params
["timeout_ns_deploy"]
1814 timeout_ns_deploy
= self
.timeout
.get("ns_deploy", self
.timeout_ns_deploy
)
1817 stage
[1] = "Getting nsr={} from db".format(nsr_id
)
1818 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
1819 stage
[1] = "Getting nsd={} from db".format(db_nsr
["nsd-id"])
1820 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
1822 # nsr_name = db_nsr["name"] # TODO short-name??
1824 # read from db: vnf's of this ns
1825 stage
[1] = "Getting vnfrs from db"
1826 self
.logger
.debug(logging_text
+ stage
[1])
1827 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
1829 # read from db: vnfd's for every vnf
1830 db_vnfds_ref
= {} # every vnfd data indexed by vnf name
1831 db_vnfds
= {} # every vnfd data indexed by vnf id
1832 db_vnfds_index
= {} # every vnfd data indexed by vnf member-index
1834 # for each vnf in ns, read vnfd
1835 for vnfr
in db_vnfrs_list
:
1836 db_vnfrs
[vnfr
["member-vnf-index-ref"]] = vnfr
# vnf's dict indexed by member-index: '1', '2', etc
1837 vnfd_id
= vnfr
["vnfd-id"] # vnfd uuid for this vnf
1838 vnfd_ref
= vnfr
["vnfd-ref"] # vnfd name for this vnf
1839 # if we haven't this vnfd, read it from db
1840 if vnfd_id
not in db_vnfds
:
1842 stage
[1] = "Getting vnfd={} id='{}' from db".format(vnfd_id
, vnfd_ref
)
1843 self
.logger
.debug(logging_text
+ stage
[1])
1844 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
1847 db_vnfds_ref
[vnfd_ref
] = vnfd
# vnfd's indexed by name
1848 db_vnfds
[vnfd_id
] = vnfd
# vnfd's indexed by id
1849 db_vnfds_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds
[vnfd_id
] # vnfd's indexed by member-index
1851 # Get or generates the _admin.deployed.VCA list
1852 vca_deployed_list
= None
1853 if db_nsr
["_admin"].get("deployed"):
1854 vca_deployed_list
= db_nsr
["_admin"]["deployed"].get("VCA")
1855 if vca_deployed_list
is None:
1856 vca_deployed_list
= []
1857 configuration_status_list
= []
1858 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1859 db_nsr_update
["configurationStatus"] = configuration_status_list
1860 # add _admin.deployed.VCA to db_nsr dictionary, value=vca_deployed_list
1861 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1862 elif isinstance(vca_deployed_list
, dict):
1863 # maintain backward compatibility. Change a dict to list at database
1864 vca_deployed_list
= list(vca_deployed_list
.values())
1865 db_nsr_update
["_admin.deployed.VCA"] = vca_deployed_list
1866 populate_dict(db_nsr
, ("_admin", "deployed", "VCA"), vca_deployed_list
)
1868 if not isinstance(deep_get(db_nsr
, ("_admin", "deployed", "RO", "vnfd")), list):
1869 populate_dict(db_nsr
, ("_admin", "deployed", "RO", "vnfd"), [])
1870 db_nsr_update
["_admin.deployed.RO.vnfd"] = []
1872 # set state to INSTANTIATED. When instantiated NBI will not delete directly
1873 db_nsr_update
["_admin.nsState"] = "INSTANTIATED"
1874 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
1876 # n2vc_redesign STEP 2 Deploy Network Scenario
1877 stage
[0] = 'Stage 2/5: deployment of KDUs, VMs and execution environments.'
1878 self
._write
_op
_status
(
1883 stage
[1] = "Deploying KDUs,"
1884 # self.logger.debug(logging_text + "Before deploy_kdus")
1885 # Call to deploy_kdus in case exists the "vdu:kdu" param
1886 await self
.deploy_kdus(
1887 logging_text
=logging_text
,
1889 nslcmop_id
=nslcmop_id
,
1892 task_instantiation_info
=tasks_dict_info
,
1895 stage
[1] = "Getting VCA public key."
1896 # n2vc_redesign STEP 1 Get VCA public ssh-key
1897 # feature 1429. Add n2vc public key to needed VMs
1898 n2vc_key
= self
.n2vc
.get_public_key()
1899 n2vc_key_list
= [n2vc_key
]
1900 if self
.vca_config
.get("public_key"):
1901 n2vc_key_list
.append(self
.vca_config
["public_key"])
1903 stage
[1] = "Deploying NS at VIM."
1904 task_ro
= asyncio
.ensure_future(
1905 self
.instantiate_RO(
1906 logging_text
=logging_text
,
1910 db_nslcmop
=db_nslcmop
,
1912 db_vnfds_ref
=db_vnfds_ref
,
1913 n2vc_key_list
=n2vc_key_list
,
1917 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_RO", task_ro
)
1918 tasks_dict_info
[task_ro
] = "Deploying at VIM"
1920 # n2vc_redesign STEP 3 to 6 Deploy N2VC
1921 stage
[1] = "Deploying Execution Environments."
1922 self
.logger
.debug(logging_text
+ stage
[1])
1924 nsi_id
= None # TODO put nsi_id when this nsr belongs to a NSI
1925 # get_iterable() returns a value from a dict or empty tuple if key does not exist
1926 for c_vnf
in get_iterable(nsd
, "constituent-vnfd"):
1927 vnfd_id
= c_vnf
["vnfd-id-ref"]
1928 vnfd
= db_vnfds_ref
[vnfd_id
]
1929 member_vnf_index
= str(c_vnf
["member-vnf-index"])
1930 db_vnfr
= db_vnfrs
[member_vnf_index
]
1931 base_folder
= vnfd
["_admin"]["storage"]
1937 # Get additional parameters
1939 if db_vnfr
.get("additionalParamsForVnf"):
1940 deploy_params
= self
._format
_additional
_params
(db_vnfr
["additionalParamsForVnf"].copy())
1942 descriptor_config
= vnfd
.get("vnf-configuration")
1943 if descriptor_config
and descriptor_config
.get("juju"):
1945 logging_text
=logging_text
+ "member_vnf_index={} ".format(member_vnf_index
),
1948 nslcmop_id
=nslcmop_id
,
1954 member_vnf_index
=member_vnf_index
,
1955 vdu_index
=vdu_index
,
1957 deploy_params
=deploy_params
,
1958 descriptor_config
=descriptor_config
,
1959 base_folder
=base_folder
,
1960 task_instantiation_info
=tasks_dict_info
,
1964 # Deploy charms for each VDU that supports one.
1965 for vdud
in get_iterable(vnfd
, 'vdu'):
1967 descriptor_config
= vdud
.get('vdu-configuration')
1968 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
1969 if vdur
.get("additionalParams"):
1970 deploy_params_vdu
= self
._format
_additional
_params
(vdur
["additionalParams"])
1972 deploy_params_vdu
= deploy_params
1973 if descriptor_config
and descriptor_config
.get("juju"):
1974 # look for vdu index in the db_vnfr["vdu"] section
1975 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
1976 # if vdur["vdu-id-ref"] == vdu_id:
1979 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
1980 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
1981 # vdu_name = vdur.get("name")
1984 for vdu_index
in range(int(vdud
.get("count", 1))):
1985 # TODO vnfr_params["rw_mgmt_ip"] = vdur["ip-address"]
1987 logging_text
=logging_text
+ "member_vnf_index={}, vdu_id={}, vdu_index={} ".format(
1988 member_vnf_index
, vdu_id
, vdu_index
),
1991 nslcmop_id
=nslcmop_id
,
1997 member_vnf_index
=member_vnf_index
,
1998 vdu_index
=vdu_index
,
2000 deploy_params
=deploy_params_vdu
,
2001 descriptor_config
=descriptor_config
,
2002 base_folder
=base_folder
,
2003 task_instantiation_info
=tasks_dict_info
,
2006 for kdud
in get_iterable(vnfd
, 'kdu'):
2007 kdu_name
= kdud
["name"]
2008 descriptor_config
= kdud
.get('kdu-configuration')
2009 if descriptor_config
and descriptor_config
.get("juju"):
2013 # look for vdu index in the db_vnfr["vdu"] section
2014 # for vdur_index, vdur in enumerate(db_vnfr["vdur"]):
2015 # if vdur["vdu-id-ref"] == vdu_id:
2018 # raise LcmException("Mismatch vdu_id={} not found in the vnfr['vdur'] list for "
2019 # "member_vnf_index={}".format(vdu_id, member_vnf_index))
2020 # vdu_name = vdur.get("name")
2024 logging_text
=logging_text
,
2027 nslcmop_id
=nslcmop_id
,
2033 member_vnf_index
=member_vnf_index
,
2034 vdu_index
=vdu_index
,
2036 deploy_params
=deploy_params
,
2037 descriptor_config
=descriptor_config
,
2038 base_folder
=base_folder
,
2039 task_instantiation_info
=tasks_dict_info
,
2043 # Check if this NS has a charm configuration
2044 descriptor_config
= nsd
.get("ns-configuration")
2045 if descriptor_config
and descriptor_config
.get("juju"):
2048 member_vnf_index
= None
2054 # Get additional parameters
2056 if db_nsr
.get("additionalParamsForNs"):
2057 deploy_params
= self
._format
_additional
_params
(db_nsr
["additionalParamsForNs"].copy())
2058 base_folder
= nsd
["_admin"]["storage"]
2060 logging_text
=logging_text
,
2063 nslcmop_id
=nslcmop_id
,
2069 member_vnf_index
=member_vnf_index
,
2070 vdu_index
=vdu_index
,
2072 deploy_params
=deploy_params
,
2073 descriptor_config
=descriptor_config
,
2074 base_folder
=base_folder
,
2075 task_instantiation_info
=tasks_dict_info
,
2079 # rest of staff will be done at finally
2081 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
2082 self
.logger
.error(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
))
2084 except asyncio
.CancelledError
:
2085 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
2086 exc
= "Operation was cancelled"
2087 except Exception as e
:
2088 exc
= traceback
.format_exc()
2089 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
2092 error_list
.append(str(exc
))
2094 # wait for pending tasks
2096 stage
[1] = "Waiting for instantiate pending tasks."
2097 self
.logger
.debug(logging_text
+ stage
[1])
2098 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_deploy
,
2099 stage
, nslcmop_id
, nsr_id
=nsr_id
)
2100 stage
[1] = stage
[2] = ""
2101 except asyncio
.CancelledError
:
2102 error_list
.append("Cancelled")
2103 # TODO cancel all tasks
2104 except Exception as exc
:
2105 error_list
.append(str(exc
))
2107 # update operation-status
2108 db_nsr_update
["operational-status"] = "running"
2109 # let's begin with VCA 'configured' status (later we can change it)
2110 db_nsr_update
["config-status"] = "configured"
2111 for task
, task_name
in tasks_dict_info
.items():
2112 if not task
.done() or task
.cancelled() or task
.exception():
2113 if task_name
.startswith(self
.task_name_deploy_vca
):
2114 # A N2VC task is pending
2115 db_nsr_update
["config-status"] = "failed"
2117 # RO or KDU task is pending
2118 db_nsr_update
["operational-status"] = "failed"
2120 # update status at database
2122 error_detail
= ". ".join(error_list
)
2123 self
.logger
.error(logging_text
+ error_detail
)
2124 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
2125 error_description_nsr
= 'Operation: INSTANTIATING.{}, Stage {}'.format(nslcmop_id
, stage
[0])
2127 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
2128 db_nslcmop_update
["detailed-status"] = error_detail
2129 nslcmop_operation_state
= "FAILED"
2133 error_description_nsr
= error_description_nslcmop
= None
2135 db_nsr_update
["detailed-status"] = "Done"
2136 db_nslcmop_update
["detailed-status"] = "Done"
2137 nslcmop_operation_state
= "COMPLETED"
2140 self
._write
_ns
_status
(
2143 current_operation
="IDLE",
2144 current_operation_id
=None,
2145 error_description
=error_description_nsr
,
2146 error_detail
=error_detail
,
2147 other_update
=db_nsr_update
2149 self
._write
_op
_status
(
2152 error_message
=error_description_nslcmop
,
2153 operation_state
=nslcmop_operation_state
,
2154 other_update
=db_nslcmop_update
,
2157 if nslcmop_operation_state
:
2159 await self
.msg
.aiowrite("ns", "instantiated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
2160 "operationState": nslcmop_operation_state
},
2162 except Exception as e
:
2163 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
2165 self
.logger
.debug(logging_text
+ "Exit")
2166 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_instantiate")
2168 async def _add_vca_relations(self
, logging_text
, nsr_id
, vca_index
: int, timeout
: int = 3600) -> bool:
2171 # 1. find all relations for this VCA
2172 # 2. wait for other peers related
2177 # STEP 1: find all relations for this VCA
2180 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2181 nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
2184 my_vca
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))[vca_index
]
2186 # read all ns-configuration relations
2187 ns_relations
= list()
2188 db_ns_relations
= deep_get(nsd
, ('ns-configuration', 'relation'))
2190 for r
in db_ns_relations
:
2191 # check if this VCA is in the relation
2192 if my_vca
.get('member-vnf-index') in\
2193 (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2194 ns_relations
.append(r
)
2196 # read all vnf-configuration relations
2197 vnf_relations
= list()
2198 db_vnfd_list
= db_nsr
.get('vnfd-id')
2200 for vnfd
in db_vnfd_list
:
2201 db_vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd
})
2202 db_vnf_relations
= deep_get(db_vnfd
, ('vnf-configuration', 'relation'))
2203 if db_vnf_relations
:
2204 for r
in db_vnf_relations
:
2205 # check if this VCA is in the relation
2206 if my_vca
.get('vdu_id') in (r
.get('entities')[0].get('id'), r
.get('entities')[1].get('id')):
2207 vnf_relations
.append(r
)
2209 # if no relations, terminate
2210 if not ns_relations
and not vnf_relations
:
2211 self
.logger
.debug(logging_text
+ ' No relations')
2214 self
.logger
.debug(logging_text
+ ' adding relations\n {}\n {}'.format(ns_relations
, vnf_relations
))
2221 if now
- start
>= timeout
:
2222 self
.logger
.error(logging_text
+ ' : timeout adding relations')
2225 # reload nsr from database (we need to update record: _admin.deloyed.VCA)
2226 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2228 # for each defined NS relation, find the VCA's related
2229 for r
in ns_relations
:
2230 from_vca_ee_id
= None
2232 from_vca_endpoint
= None
2233 to_vca_endpoint
= None
2234 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2235 for vca
in vca_list
:
2236 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id') \
2237 and vca
.get('config_sw_installed'):
2238 from_vca_ee_id
= vca
.get('ee_id')
2239 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2240 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id') \
2241 and vca
.get('config_sw_installed'):
2242 to_vca_ee_id
= vca
.get('ee_id')
2243 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2244 if from_vca_ee_id
and to_vca_ee_id
:
2246 await self
.n2vc
.add_relation(
2247 ee_id_1
=from_vca_ee_id
,
2248 ee_id_2
=to_vca_ee_id
,
2249 endpoint_1
=from_vca_endpoint
,
2250 endpoint_2
=to_vca_endpoint
)
2251 # remove entry from relations list
2252 ns_relations
.remove(r
)
2254 # check failed peers
2256 vca_status_list
= db_nsr
.get('configurationStatus')
2258 for i
in range(len(vca_list
)):
2260 vca_status
= vca_status_list
[i
]
2261 if vca
.get('member-vnf-index') == r
.get('entities')[0].get('id'):
2262 if vca_status
.get('status') == 'BROKEN':
2263 # peer broken: remove relation from list
2264 ns_relations
.remove(r
)
2265 if vca
.get('member-vnf-index') == r
.get('entities')[1].get('id'):
2266 if vca_status
.get('status') == 'BROKEN':
2267 # peer broken: remove relation from list
2268 ns_relations
.remove(r
)
2273 # for each defined VNF relation, find the VCA's related
2274 for r
in vnf_relations
:
2275 from_vca_ee_id
= None
2277 from_vca_endpoint
= None
2278 to_vca_endpoint
= None
2279 vca_list
= deep_get(db_nsr
, ('_admin', 'deployed', 'VCA'))
2280 for vca
in vca_list
:
2281 if vca
.get('vdu_id') == r
.get('entities')[0].get('id') and vca
.get('config_sw_installed'):
2282 from_vca_ee_id
= vca
.get('ee_id')
2283 from_vca_endpoint
= r
.get('entities')[0].get('endpoint')
2284 if vca
.get('vdu_id') == r
.get('entities')[1].get('id') and vca
.get('config_sw_installed'):
2285 to_vca_ee_id
= vca
.get('ee_id')
2286 to_vca_endpoint
= r
.get('entities')[1].get('endpoint')
2287 if from_vca_ee_id
and to_vca_ee_id
:
2289 await self
.n2vc
.add_relation(
2290 ee_id_1
=from_vca_ee_id
,
2291 ee_id_2
=to_vca_ee_id
,
2292 endpoint_1
=from_vca_endpoint
,
2293 endpoint_2
=to_vca_endpoint
)
2294 # remove entry from relations list
2295 vnf_relations
.remove(r
)
2297 # check failed peers
2299 vca_status_list
= db_nsr
.get('configurationStatus')
2301 for i
in range(len(vca_list
)):
2303 vca_status
= vca_status_list
[i
]
2304 if vca
.get('vdu_id') == r
.get('entities')[0].get('id'):
2305 if vca_status
.get('status') == 'BROKEN':
2306 # peer broken: remove relation from list
2307 ns_relations
.remove(r
)
2308 if vca
.get('vdu_id') == r
.get('entities')[1].get('id'):
2309 if vca_status
.get('status') == 'BROKEN':
2310 # peer broken: remove relation from list
2311 ns_relations
.remove(r
)
2317 await asyncio
.sleep(5.0)
2319 if not ns_relations
and not vnf_relations
:
2320 self
.logger
.debug('Relations added')
2325 except Exception as e
:
2326 self
.logger
.warn(logging_text
+ ' ERROR adding relations: {}'.format(e
))
2329 def _write_db_callback(self
, task
, item
, _id
, on_done
=None, on_exc
=None):
2331 callback for kdu install intended to store the returned kdu_instance at database
2336 result
= task
.result()
2338 db_update
[on_done
] = str(result
)
2339 except Exception as e
:
2341 db_update
[on_exc
] = str(e
)
2344 self
.update_db_2(item
, _id
, db_update
)
2348 async def deploy_kdus(self
, logging_text
, nsr_id
, nslcmop_id
, db_vnfrs
, db_vnfds
, task_instantiation_info
):
2349 # Launch kdus if present in the descriptor
2351 k8scluster_id_2_uuic
= {"helm-chart": {}, "juju-bundle": {}}
2353 def _get_cluster_id(cluster_id
, cluster_type
):
2354 nonlocal k8scluster_id_2_uuic
2355 if cluster_id
in k8scluster_id_2_uuic
[cluster_type
]:
2356 return k8scluster_id_2_uuic
[cluster_type
][cluster_id
]
2358 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": cluster_id
}, fail_on_empty
=False)
2359 if not db_k8scluster
:
2360 raise LcmException("K8s cluster {} cannot be found".format(cluster_id
))
2361 k8s_id
= deep_get(db_k8scluster
, ("_admin", cluster_type
, "id"))
2363 raise LcmException("K8s cluster '{}' has not been initilized for '{}'".format(cluster_id
, cluster_type
))
2364 k8scluster_id_2_uuic
[cluster_type
][cluster_id
] = k8s_id
2367 logging_text
+= "Deploy kdus: "
2370 db_nsr_update
= {"_admin.deployed.K8s": []}
2371 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2374 updated_cluster_list
= []
2376 for vnfr_data
in db_vnfrs
.values():
2377 for kdur
in get_iterable(vnfr_data
, "kdur"):
2378 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
2379 vnfd_id
= vnfr_data
.get('vnfd-id')
2380 namespace
= kdur
.get("k8s-namespace")
2381 if kdur
.get("helm-chart"):
2382 kdumodel
= kdur
["helm-chart"]
2383 k8sclustertype
= "helm-chart"
2384 elif kdur
.get("juju-bundle"):
2385 kdumodel
= kdur
["juju-bundle"]
2386 k8sclustertype
= "juju-bundle"
2388 raise LcmException("kdu type for kdu='{}.{}' is neither helm-chart nor "
2389 "juju-bundle. Maybe an old NBI version is running".
2390 format(vnfr_data
["member-vnf-index-ref"], kdur
["kdu-name"]))
2391 # check if kdumodel is a file and exists
2393 storage
= deep_get(db_vnfds
.get(vnfd_id
), ('_admin', 'storage'))
2394 if storage
and storage
.get('pkg-dir'): # may be not present if vnfd has not artifacts
2395 # path format: /vnfdid/pkkdir/helm-charts|juju-bundles/kdumodel
2396 filename
= '{}/{}/{}s/{}'.format(storage
["folder"], storage
["pkg-dir"], k8sclustertype
,
2398 if self
.fs
.file_exists(filename
, mode
='file') or self
.fs
.file_exists(filename
, mode
='dir'):
2399 kdumodel
= self
.fs
.path
+ filename
2400 except (asyncio
.TimeoutError
, asyncio
.CancelledError
):
2402 except Exception: # it is not a file
2405 k8s_cluster_id
= kdur
["k8s-cluster"]["id"]
2406 step
= "Synchronize repos for k8s cluster '{}'".format(k8s_cluster_id
)
2407 cluster_uuid
= _get_cluster_id(k8s_cluster_id
, k8sclustertype
)
2409 if k8sclustertype
== "helm-chart" and cluster_uuid
not in updated_cluster_list
:
2410 del_repo_list
, added_repo_dict
= await asyncio
.ensure_future(
2411 self
.k8sclusterhelm
.synchronize_repos(cluster_uuid
=cluster_uuid
))
2412 if del_repo_list
or added_repo_dict
:
2413 unset
= {'_admin.helm_charts_added.' + item
: None for item
in del_repo_list
}
2414 updated
= {'_admin.helm_charts_added.' +
2415 item
: name
for item
, name
in added_repo_dict
.items()}
2416 self
.logger
.debug(logging_text
+ "repos synchronized on k8s cluster '{}' to_delete: {}, "
2417 "to_add: {}".format(k8s_cluster_id
, del_repo_list
,
2419 self
.db
.set_one("k8sclusters", {"_id": k8s_cluster_id
}, updated
, unset
=unset
)
2420 updated_cluster_list
.append(cluster_uuid
)
2422 step
= "Instantiating KDU {}.{} in k8s cluster {}".format(vnfr_data
["member-vnf-index-ref"],
2423 kdur
["kdu-name"], k8s_cluster_id
)
2425 k8s_instace_info
= {"kdu-instance": None,
2426 "k8scluster-uuid": cluster_uuid
,
2427 "k8scluster-type": k8sclustertype
,
2428 "member-vnf-index": vnfr_data
["member-vnf-index-ref"],
2429 "kdu-name": kdur
["kdu-name"],
2430 "kdu-model": kdumodel
,
2431 "namespace": namespace
}
2432 db_path
= "_admin.deployed.K8s.{}".format(index
)
2433 db_nsr_update
[db_path
] = k8s_instace_info
2434 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2436 db_dict
= {"collection": "nsrs",
2437 "filter": {"_id": nsr_id
},
2440 task
= asyncio
.ensure_future(
2441 self
.k8scluster_map
[k8sclustertype
].install(cluster_uuid
=cluster_uuid
, kdu_model
=kdumodel
,
2442 atomic
=True, params
=desc_params
,
2443 db_dict
=db_dict
, timeout
=600,
2444 kdu_name
=kdur
["kdu-name"], namespace
=namespace
))
2446 task
.add_done_callback(partial(self
._write
_db
_callback
, item
="nsrs", _id
=nsr_id
,
2447 on_done
=db_path
+ ".kdu-instance",
2448 on_exc
=db_path
+ ".detailed-status"))
2449 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_KDU-{}".format(index
), task
)
2450 task_instantiation_info
[task
] = "Deploying KDU {}".format(kdur
["kdu-name"])
2454 except (LcmException
, asyncio
.CancelledError
):
2456 except Exception as e
:
2457 msg
= "Exception {} while {}: {}".format(type(e
).__name
__, step
, e
)
2458 if isinstance(e
, (N2VCException
, DbException
)):
2459 self
.logger
.error(logging_text
+ msg
)
2461 self
.logger
.critical(logging_text
+ msg
, exc_info
=True)
2462 raise LcmException(msg
)
2465 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2467 def _deploy_n2vc(self
, logging_text
, db_nsr
, db_vnfr
, nslcmop_id
, nsr_id
, nsi_id
, vnfd_id
, vdu_id
,
2468 kdu_name
, member_vnf_index
, vdu_index
, vdu_name
, deploy_params
, descriptor_config
,
2469 base_folder
, task_instantiation_info
, stage
):
2470 # launch instantiate_N2VC in a asyncio task and register task object
2471 # Look where information of this charm is at database <nsrs>._admin.deployed.VCA
2472 # if not found, create one entry and update database
2474 # fill db_nsr._admin.deployed.VCA.<index>
2476 for vca_index
, vca_deployed
in enumerate(db_nsr
["_admin"]["deployed"]["VCA"]):
2477 if not vca_deployed
:
2479 if vca_deployed
.get("member-vnf-index") == member_vnf_index
and \
2480 vca_deployed
.get("vdu_id") == vdu_id
and \
2481 vca_deployed
.get("kdu_name") == kdu_name
and \
2482 vca_deployed
.get("vdu_count_index", 0) == vdu_index
:
2485 # not found, create one.
2487 "member-vnf-index": member_vnf_index
,
2489 "kdu_name": kdu_name
,
2490 "vdu_count_index": vdu_index
,
2491 "operational-status": "init", # TODO revise
2492 "detailed-status": "", # TODO revise
2493 "step": "initial-deploy", # TODO revise
2495 "vdu_name": vdu_name
,
2499 # create VCA and configurationStatus in db
2501 "_admin.deployed.VCA.{}".format(vca_index
): vca_deployed
,
2502 "configurationStatus.{}".format(vca_index
): dict()
2504 self
.update_db_2("nsrs", nsr_id
, db_dict
)
2506 db_nsr
["_admin"]["deployed"]["VCA"].append(vca_deployed
)
2509 task_n2vc
= asyncio
.ensure_future(
2510 self
.instantiate_N2VC(
2511 logging_text
=logging_text
,
2512 vca_index
=vca_index
,
2518 vdu_index
=vdu_index
,
2519 deploy_params
=deploy_params
,
2520 config_descriptor
=descriptor_config
,
2521 base_folder
=base_folder
,
2522 nslcmop_id
=nslcmop_id
,
2526 self
.lcm_tasks
.register("ns", nsr_id
, nslcmop_id
, "instantiate_N2VC-{}".format(vca_index
), task_n2vc
)
2527 task_instantiation_info
[task_n2vc
] = self
.task_name_deploy_vca
+ " {}.{}".format(
2528 member_vnf_index
or "", vdu_id
or "")
2530 # Check if this VNFD has a configured terminate action
2531 def _has_terminate_config_primitive(self
, vnfd
):
2532 vnf_config
= vnfd
.get("vnf-configuration")
2533 if vnf_config
and vnf_config
.get("terminate-config-primitive"):
2539 def _get_terminate_config_primitive_seq_list(vnfd
):
2540 """ Get a numerically sorted list of the sequences for this VNFD's terminate action """
2541 # No need to check for existing primitive twice, already done before
2542 vnf_config
= vnfd
.get("vnf-configuration")
2543 seq_list
= vnf_config
.get("terminate-config-primitive")
2544 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2545 seq_list_sorted
= sorted(seq_list
, key
=lambda x
: int(x
['seq']))
2546 return seq_list_sorted
2549 def _create_nslcmop(nsr_id
, operation
, params
):
2551 Creates a ns-lcm-opp content to be stored at database.
2552 :param nsr_id: internal id of the instance
2553 :param operation: instantiate, terminate, scale, action, ...
2554 :param params: user parameters for the operation
2555 :return: dictionary following SOL005 format
2557 # Raise exception if invalid arguments
2558 if not (nsr_id
and operation
and params
):
2560 "Parameters 'nsr_id', 'operation' and 'params' needed to create primitive not provided")
2566 # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
2567 "operationState": "PROCESSING",
2568 "statusEnteredTime": now
,
2569 "nsInstanceId": nsr_id
,
2570 "lcmOperationType": operation
,
2572 "isAutomaticInvocation": False,
2573 "operationParams": params
,
2574 "isCancelPending": False,
2576 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
2577 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsr_id
,
2582 def _format_additional_params(self
, params
):
2583 params
= params
or {}
2584 for key
, value
in params
.items():
2585 if str(value
).startswith("!!yaml "):
2586 params
[key
] = yaml
.safe_load(value
[7:])
2589 def _get_terminate_primitive_params(self
, seq
, vnf_index
):
2590 primitive
= seq
.get('name')
2591 primitive_params
= {}
2593 "member_vnf_index": vnf_index
,
2594 "primitive": primitive
,
2595 "primitive_params": primitive_params
,
2598 return self
._map
_primitive
_params
(seq
, params
, desc_params
)
2602 def _retry_or_skip_suboperation(self
, db_nslcmop
, op_index
):
2603 op
= deep_get(db_nslcmop
, ('_admin', 'operations'), [])[op_index
]
2604 if op
.get('operationState') == 'COMPLETED':
2605 # b. Skip sub-operation
2606 # _ns_execute_primitive() or RO.create_action() will NOT be executed
2607 return self
.SUBOPERATION_STATUS_SKIP
2609 # c. retry executing sub-operation
2610 # The sub-operation exists, and operationState != 'COMPLETED'
2611 # Update operationState = 'PROCESSING' to indicate a retry.
2612 operationState
= 'PROCESSING'
2613 detailed_status
= 'In progress'
2614 self
._update
_suboperation
_status
(
2615 db_nslcmop
, op_index
, operationState
, detailed_status
)
2616 # Return the sub-operation index
2617 # _ns_execute_primitive() or RO.create_action() will be called from scale()
2618 # with arguments extracted from the sub-operation
2621 # Find a sub-operation where all keys in a matching dictionary must match
2622 # Returns the index of the matching sub-operation, or SUBOPERATION_STATUS_NOT_FOUND if no match
2623 def _find_suboperation(self
, db_nslcmop
, match
):
2624 if db_nslcmop
and match
:
2625 op_list
= db_nslcmop
.get('_admin', {}).get('operations', [])
2626 for i
, op
in enumerate(op_list
):
2627 if all(op
.get(k
) == match
[k
] for k
in match
):
2629 return self
.SUBOPERATION_STATUS_NOT_FOUND
2631 # Update status for a sub-operation given its index
2632 def _update_suboperation_status(self
, db_nslcmop
, op_index
, operationState
, detailed_status
):
2633 # Update DB for HA tasks
2634 q_filter
= {'_id': db_nslcmop
['_id']}
2635 update_dict
= {'_admin.operations.{}.operationState'.format(op_index
): operationState
,
2636 '_admin.operations.{}.detailed-status'.format(op_index
): detailed_status
}
2637 self
.db
.set_one("nslcmops",
2639 update_dict
=update_dict
,
2640 fail_on_empty
=False)
2642 # Add sub-operation, return the index of the added sub-operation
2643 # Optionally, set operationState, detailed-status, and operationType
2644 # Status and type are currently set for 'scale' sub-operations:
2645 # 'operationState' : 'PROCESSING' | 'COMPLETED' | 'FAILED'
2646 # 'detailed-status' : status message
2647 # 'operationType': may be any type, in the case of scaling: 'PRE-SCALE' | 'POST-SCALE'
2648 # Status and operation type are currently only used for 'scale', but NOT for 'terminate' sub-operations.
2649 def _add_suboperation(self
, db_nslcmop
, vnf_index
, vdu_id
, vdu_count_index
, vdu_name
, primitive
,
2650 mapped_primitive_params
, operationState
=None, detailed_status
=None, operationType
=None,
2651 RO_nsr_id
=None, RO_scaling_info
=None):
2653 return self
.SUBOPERATION_STATUS_NOT_FOUND
2654 # Get the "_admin.operations" list, if it exists
2655 db_nslcmop_admin
= db_nslcmop
.get('_admin', {})
2656 op_list
= db_nslcmop_admin
.get('operations')
2657 # Create or append to the "_admin.operations" list
2658 new_op
= {'member_vnf_index': vnf_index
,
2660 'vdu_count_index': vdu_count_index
,
2661 'primitive': primitive
,
2662 'primitive_params': mapped_primitive_params
}
2664 new_op
['operationState'] = operationState
2666 new_op
['detailed-status'] = detailed_status
2668 new_op
['lcmOperationType'] = operationType
2670 new_op
['RO_nsr_id'] = RO_nsr_id
2672 new_op
['RO_scaling_info'] = RO_scaling_info
2674 # No existing operations, create key 'operations' with current operation as first list element
2675 db_nslcmop_admin
.update({'operations': [new_op
]})
2676 op_list
= db_nslcmop_admin
.get('operations')
2678 # Existing operations, append operation to list
2679 op_list
.append(new_op
)
2681 db_nslcmop_update
= {'_admin.operations': op_list
}
2682 self
.update_db_2("nslcmops", db_nslcmop
['_id'], db_nslcmop_update
)
2683 op_index
= len(op_list
) - 1
2686 # Helper methods for scale() sub-operations
2688 # pre-scale/post-scale:
2689 # Check for 3 different cases:
2690 # a. New: First time execution, return SUBOPERATION_STATUS_NEW
2691 # b. Skip: Existing sub-operation exists, operationState == 'COMPLETED', return SUBOPERATION_STATUS_SKIP
2692 # c. retry: Existing sub-operation exists, operationState != 'COMPLETED', return op_index to re-execute
2693 def _check_or_add_scale_suboperation(self
, db_nslcmop
, vnf_index
, vnf_config_primitive
, primitive_params
,
2694 operationType
, RO_nsr_id
=None, RO_scaling_info
=None):
2695 # Find this sub-operation
2696 if RO_nsr_id
and RO_scaling_info
:
2697 operationType
= 'SCALE-RO'
2699 'member_vnf_index': vnf_index
,
2700 'RO_nsr_id': RO_nsr_id
,
2701 'RO_scaling_info': RO_scaling_info
,
2705 'member_vnf_index': vnf_index
,
2706 'primitive': vnf_config_primitive
,
2707 'primitive_params': primitive_params
,
2708 'lcmOperationType': operationType
2710 op_index
= self
._find
_suboperation
(db_nslcmop
, match
)
2711 if op_index
== self
.SUBOPERATION_STATUS_NOT_FOUND
:
2712 # a. New sub-operation
2713 # The sub-operation does not exist, add it.
2714 # _ns_execute_primitive() will be called from scale() as usual, with non-modified arguments
2715 # The following parameters are set to None for all kind of scaling:
2717 vdu_count_index
= None
2719 if RO_nsr_id
and RO_scaling_info
:
2720 vnf_config_primitive
= None
2721 primitive_params
= None
2724 RO_scaling_info
= None
2725 # Initial status for sub-operation
2726 operationState
= 'PROCESSING'
2727 detailed_status
= 'In progress'
2728 # Add sub-operation for pre/post-scaling (zero or more operations)
2729 self
._add
_suboperation
(db_nslcmop
,
2734 vnf_config_primitive
,
2741 return self
.SUBOPERATION_STATUS_NEW
2743 # Return either SUBOPERATION_STATUS_SKIP (operationState == 'COMPLETED'),
2744 # or op_index (operationState != 'COMPLETED')
2745 return self
._retry
_or
_skip
_suboperation
(db_nslcmop
, op_index
)
2747 # Function to return execution_environment id
2749 def _get_ee_id(self
, vnf_index
, vdu_id
, vca_deployed_list
):
2750 # TODO vdu_index_count
2751 for vca
in vca_deployed_list
:
2752 if vca
["member-vnf-index"] == vnf_index
and vca
["vdu_id"] == vdu_id
:
2755 async def destroy_N2VC(self
, logging_text
, db_nslcmop
, vca_deployed
, config_descriptor
, vca_index
, destroy_ee
=True):
2757 Execute the terminate primitives and destroy the execution environment (if destroy_ee=False
2758 :param logging_text:
2760 :param vca_deployed: Dictionary of deployment info at db_nsr._admin.depoloyed.VCA.<INDEX>
2761 :param config_descriptor: Configuration descriptor of the NSD, VNFD, VNFD.vdu or VNFD.kdu
2762 :param vca_index: index in the database _admin.deployed.VCA
2763 :param destroy_ee: False to do not destroy, because it will be destroyed all of then at once
2764 :return: None or exception
2766 # execute terminate_primitives
2767 terminate_primitives
= config_descriptor
.get("terminate-config-primitive")
2768 vdu_id
= vca_deployed
.get("vdu_id")
2769 vdu_count_index
= vca_deployed
.get("vdu_count_index")
2770 vdu_name
= vca_deployed
.get("vdu_name")
2771 vnf_index
= vca_deployed
.get("member-vnf-index")
2772 if terminate_primitives
and vca_deployed
.get("needed_terminate"):
2773 # Get all 'seq' tags in seq_list, order sequences numerically, ascending.
2774 terminate_primitives
= sorted(terminate_primitives
, key
=lambda x
: int(x
['seq']))
2775 for seq
in terminate_primitives
:
2776 # For each sequence in list, get primitive and call _ns_execute_primitive()
2777 step
= "Calling terminate action for vnf_member_index={} primitive={}".format(
2778 vnf_index
, seq
.get("name"))
2779 self
.logger
.debug(logging_text
+ step
)
2780 # Create the primitive for each sequence, i.e. "primitive": "touch"
2781 primitive
= seq
.get('name')
2782 mapped_primitive_params
= self
._get
_terminate
_primitive
_params
(seq
, vnf_index
)
2783 # The following 3 parameters are currently set to None for 'terminate':
2784 # vdu_id, vdu_count_index, vdu_name
2787 self
._add
_suboperation
(db_nslcmop
,
2793 mapped_primitive_params
)
2794 # Sub-operations: Call _ns_execute_primitive() instead of action()
2796 result
, result_detail
= await self
._ns
_execute
_primitive
(vca_deployed
["ee_id"], primitive
,
2797 mapped_primitive_params
)
2798 except LcmException
:
2799 # this happens when VCA is not deployed. In this case it is not needed to terminate
2801 result_ok
= ['COMPLETED', 'PARTIALLY_COMPLETED']
2802 if result
not in result_ok
:
2803 raise LcmException("terminate_primitive {} for vnf_member_index={} fails with "
2804 "error {}".format(seq
.get("name"), vnf_index
, result_detail
))
2805 # set that this VCA do not need terminated
2806 db_update_entry
= "_admin.deployed.VCA.{}.needed_terminate".format(vca_index
)
2807 self
.update_db_2("nsrs", db_nslcmop
["nsInstanceId"], {db_update_entry
: False})
2810 await self
.n2vc
.delete_execution_environment(vca_deployed
["ee_id"])
2812 async def _delete_all_N2VC(self
, db_nsr
: dict):
2813 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='TERMINATING')
2814 namespace
= "." + db_nsr
["_id"]
2816 await self
.n2vc
.delete_namespace(namespace
=namespace
, total_timeout
=self
.timeout_charm_delete
)
2817 except N2VCNotFound
: # already deleted. Skip
2819 self
._write
_all
_config
_status
(db_nsr
=db_nsr
, status
='DELETED')
2821 async def _terminate_RO(self
, logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
):
2823 Terminates a deployment from RO
2824 :param logging_text:
2825 :param nsr_deployed: db_nsr._admin.deployed
2828 :param stage: list of string with the content to write on db_nslcmop.detailed-status.
2829 this method will update only the index 2, but it will write on database the concatenated content of the list
2834 ro_nsr_id
= ro_delete_action
= None
2835 if nsr_deployed
and nsr_deployed
.get("RO"):
2836 ro_nsr_id
= nsr_deployed
["RO"].get("nsr_id")
2837 ro_delete_action
= nsr_deployed
["RO"].get("nsr_delete_action_id")
2840 stage
[2] = "Deleting ns from VIM."
2841 db_nsr_update
["detailed-status"] = " ".join(stage
)
2842 self
._write
_op
_status
(nslcmop_id
, stage
)
2843 self
.logger
.debug(logging_text
+ stage
[2])
2844 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2845 self
._write
_op
_status
(nslcmop_id
, stage
)
2846 desc
= await self
.RO
.delete("ns", ro_nsr_id
)
2847 ro_delete_action
= desc
["action_id"]
2848 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = ro_delete_action
2849 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2850 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2851 if ro_delete_action
:
2852 # wait until NS is deleted from VIM
2853 stage
[2] = "Waiting ns deleted from VIM."
2854 detailed_status_old
= None
2855 self
.logger
.debug(logging_text
+ stage
[2] + " RO_id={} ro_delete_action={}".format(ro_nsr_id
,
2857 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2858 self
._write
_op
_status
(nslcmop_id
, stage
)
2860 delete_timeout
= 20 * 60 # 20 minutes
2861 while delete_timeout
> 0:
2862 desc
= await self
.RO
.show(
2864 item_id_name
=ro_nsr_id
,
2865 extra_item
="action",
2866 extra_item_id
=ro_delete_action
)
2869 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
2871 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
2872 if ns_status
== "ERROR":
2873 raise ROclient
.ROClientException(ns_status_info
)
2874 elif ns_status
== "BUILD":
2875 stage
[2] = "Deleting from VIM {}".format(ns_status_info
)
2876 elif ns_status
== "ACTIVE":
2877 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2878 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2881 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
2882 if stage
[2] != detailed_status_old
:
2883 detailed_status_old
= stage
[2]
2884 db_nsr_update
["detailed-status"] = " ".join(stage
)
2885 self
._write
_op
_status
(nslcmop_id
, stage
)
2886 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2887 await asyncio
.sleep(5, loop
=self
.loop
)
2889 else: # delete_timeout <= 0:
2890 raise ROclient
.ROClientException("Timeout waiting ns deleted from VIM")
2892 except Exception as e
:
2893 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2894 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2895 db_nsr_update
["_admin.deployed.RO.nsr_id"] = None
2896 db_nsr_update
["_admin.deployed.RO.nsr_status"] = "DELETED"
2897 db_nsr_update
["_admin.deployed.RO.nsr_delete_action_id"] = None
2898 self
.logger
.debug(logging_text
+ "RO_ns_id={} already deleted".format(ro_nsr_id
))
2899 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2900 failed_detail
.append("delete conflict: {}".format(e
))
2901 self
.logger
.debug(logging_text
+ "RO_ns_id={} delete conflict: {}".format(ro_nsr_id
, e
))
2903 failed_detail
.append("delete error: {}".format(e
))
2904 self
.logger
.error(logging_text
+ "RO_ns_id={} delete error: {}".format(ro_nsr_id
, e
))
2907 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "nsd_id")):
2908 ro_nsd_id
= nsr_deployed
["RO"]["nsd_id"]
2910 stage
[2] = "Deleting nsd from RO."
2911 db_nsr_update
["detailed-status"] = " ".join(stage
)
2912 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2913 self
._write
_op
_status
(nslcmop_id
, stage
)
2914 await self
.RO
.delete("nsd", ro_nsd_id
)
2915 self
.logger
.debug(logging_text
+ "ro_nsd_id={} deleted".format(ro_nsd_id
))
2916 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2917 except Exception as e
:
2918 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2919 db_nsr_update
["_admin.deployed.RO.nsd_id"] = None
2920 self
.logger
.debug(logging_text
+ "ro_nsd_id={} already deleted".format(ro_nsd_id
))
2921 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2922 failed_detail
.append("ro_nsd_id={} delete conflict: {}".format(ro_nsd_id
, e
))
2923 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2925 failed_detail
.append("ro_nsd_id={} delete error: {}".format(ro_nsd_id
, e
))
2926 self
.logger
.error(logging_text
+ failed_detail
[-1])
2928 if not failed_detail
and deep_get(nsr_deployed
, ("RO", "vnfd")):
2929 for index
, vnf_deployed
in enumerate(nsr_deployed
["RO"]["vnfd"]):
2930 if not vnf_deployed
or not vnf_deployed
["id"]:
2933 ro_vnfd_id
= vnf_deployed
["id"]
2934 stage
[2] = "Deleting member_vnf_index={} ro_vnfd_id={} from RO.".format(
2935 vnf_deployed
["member-vnf-index"], ro_vnfd_id
)
2936 db_nsr_update
["detailed-status"] = " ".join(stage
)
2937 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2938 self
._write
_op
_status
(nslcmop_id
, stage
)
2939 await self
.RO
.delete("vnfd", ro_vnfd_id
)
2940 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} deleted".format(ro_vnfd_id
))
2941 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2942 except Exception as e
:
2943 if isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 404: # not found
2944 db_nsr_update
["_admin.deployed.RO.vnfd.{}.id".format(index
)] = None
2945 self
.logger
.debug(logging_text
+ "ro_vnfd_id={} already deleted ".format(ro_vnfd_id
))
2946 elif isinstance(e
, ROclient
.ROClientException
) and e
.http_code
== 409: # conflict
2947 failed_detail
.append("ro_vnfd_id={} delete conflict: {}".format(ro_vnfd_id
, e
))
2948 self
.logger
.debug(logging_text
+ failed_detail
[-1])
2950 failed_detail
.append("ro_vnfd_id={} delete error: {}".format(ro_vnfd_id
, e
))
2951 self
.logger
.error(logging_text
+ failed_detail
[-1])
2954 stage
[2] = "Error deleting from VIM"
2956 stage
[2] = "Deleted from VIM"
2957 db_nsr_update
["detailed-status"] = " ".join(stage
)
2958 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
2959 self
._write
_op
_status
(nslcmop_id
, stage
)
2962 raise LcmException("; ".join(failed_detail
))
2964 async def terminate(self
, nsr_id
, nslcmop_id
):
2965 # Try to lock HA task here
2966 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
2967 if not task_is_locked_by_me
:
2970 logging_text
= "Task ns={} terminate={} ".format(nsr_id
, nslcmop_id
)
2971 self
.logger
.debug(logging_text
+ "Enter")
2972 timeout_ns_terminate
= self
.timeout_ns_terminate
2975 operation_params
= None
2977 error_list
= [] # annotates all failed error messages
2978 db_nslcmop_update
= {}
2979 autoremove
= False # autoremove after terminated
2980 tasks_dict_info
= {}
2982 stage
= ["Stage 1/3: Preparing task.", "Waiting for previous operations to terminate.", ""]
2983 # ^ contains [stage, step, VIM-status]
2985 # wait for any previous tasks in process
2986 await self
.lcm_tasks
.waitfor_related_HA("ns", 'nslcmops', nslcmop_id
)
2988 stage
[1] = "Getting nslcmop={} from db.".format(nslcmop_id
)
2989 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
2990 operation_params
= db_nslcmop
.get("operationParams") or {}
2991 if operation_params
.get("timeout_ns_terminate"):
2992 timeout_ns_terminate
= operation_params
["timeout_ns_terminate"]
2993 stage
[1] = "Getting nsr={} from db.".format(nsr_id
)
2994 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2996 db_nsr_update
["operational-status"] = "terminating"
2997 db_nsr_update
["config-status"] = "terminating"
2998 self
._write
_ns
_status
(
3000 ns_state
="TERMINATING",
3001 current_operation
="TERMINATING",
3002 current_operation_id
=nslcmop_id
,
3003 other_update
=db_nsr_update
3005 self
._write
_op
_status
(
3010 nsr_deployed
= deepcopy(db_nsr
["_admin"].get("deployed")) or {}
3011 if db_nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
3014 stage
[1] = "Getting vnf descriptors from db."
3015 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
3016 db_vnfds_from_id
= {}
3017 db_vnfds_from_member_index
= {}
3019 for vnfr
in db_vnfrs_list
:
3020 vnfd_id
= vnfr
["vnfd-id"]
3021 if vnfd_id
not in db_vnfds_from_id
:
3022 vnfd
= self
.db
.get_one("vnfds", {"_id": vnfd_id
})
3023 db_vnfds_from_id
[vnfd_id
] = vnfd
3024 db_vnfds_from_member_index
[vnfr
["member-vnf-index-ref"]] = db_vnfds_from_id
[vnfd_id
]
3026 # Destroy individual execution environments when there are terminating primitives.
3027 # Rest of EE will be deleted at once
3028 if not operation_params
.get("skip_terminate_primitives"):
3029 stage
[0] = "Stage 2/3 execute terminating primitives."
3030 stage
[1] = "Looking execution environment that needs terminate."
3031 self
.logger
.debug(logging_text
+ stage
[1])
3032 for vca_index
, vca
in enumerate(get_iterable(nsr_deployed
, "VCA")):
3033 config_descriptor
= None
3034 if not vca
or not vca
.get("ee_id") or not vca
.get("needed_terminate"):
3036 if not vca
.get("member-vnf-index"):
3038 config_descriptor
= db_nsr
.get("ns-configuration")
3039 elif vca
.get("vdu_id"):
3040 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3041 vdud
= next((vdu
for vdu
in db_vnfd
.get("vdu", ()) if vdu
["id"] == vca
.get("vdu_id")), None)
3043 config_descriptor
= vdud
.get("vdu-configuration")
3044 elif vca
.get("kdu_name"):
3045 db_vnfd
= db_vnfds_from_member_index
[vca
["member-vnf-index"]]
3046 kdud
= next((kdu
for kdu
in db_vnfd
.get("kdu", ()) if kdu
["name"] == vca
.get("kdu_name")), None)
3048 config_descriptor
= kdud
.get("kdu-configuration")
3050 config_descriptor
= db_vnfds_from_member_index
[vca
["member-vnf-index"]].get("vnf-configuration")
3051 task
= asyncio
.ensure_future(self
.destroy_N2VC(logging_text
, db_nslcmop
, vca
, config_descriptor
,
3053 tasks_dict_info
[task
] = "Terminating VCA {}".format(vca
.get("ee_id"))
3055 # wait for pending tasks of terminate primitives
3057 self
.logger
.debug(logging_text
+ 'Waiting for terminate primitive pending tasks...')
3058 error_list
= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
,
3059 min(self
.timeout_charm_delete
, timeout_ns_terminate
),
3062 return # raise LcmException("; ".join(error_list))
3063 tasks_dict_info
.clear()
3065 # remove All execution environments at once
3066 stage
[0] = "Stage 3/3 delete all."
3068 if nsr_deployed
.get("VCA"):
3069 stage
[1] = "Deleting all execution environments."
3070 self
.logger
.debug(logging_text
+ stage
[1])
3071 task_delete_ee
= asyncio
.ensure_future(asyncio
.wait_for(self
._delete
_all
_N
2VC
(db_nsr
=db_nsr
),
3072 timeout
=self
.timeout_charm_delete
))
3073 # task_delete_ee = asyncio.ensure_future(self.n2vc.delete_namespace(namespace="." + nsr_id))
3074 tasks_dict_info
[task_delete_ee
] = "Terminating all VCA"
3076 # Delete from k8scluster
3077 stage
[1] = "Deleting KDUs."
3078 self
.logger
.debug(logging_text
+ stage
[1])
3079 # print(nsr_deployed)
3080 for kdu
in get_iterable(nsr_deployed
, "K8s"):
3081 if not kdu
or not kdu
.get("kdu-instance"):
3083 kdu_instance
= kdu
.get("kdu-instance")
3084 if kdu
.get("k8scluster-type") in self
.k8scluster_map
:
3085 task_delete_kdu_instance
= asyncio
.ensure_future(
3086 self
.k8scluster_map
[kdu
["k8scluster-type"]].uninstall(
3087 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3088 kdu_instance
=kdu_instance
))
3090 self
.logger
.error(logging_text
+ "Unknown k8s deployment type {}".
3091 format(kdu
.get("k8scluster-type")))
3093 tasks_dict_info
[task_delete_kdu_instance
] = "Terminating KDU '{}'".format(kdu
.get("kdu-name"))
3096 stage
[1] = "Deleting ns from VIM."
3098 task_delete_ro
= asyncio
.ensure_future(
3099 self
._terminate
_ng
_ro
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3101 task_delete_ro
= asyncio
.ensure_future(
3102 self
._terminate
_RO
(logging_text
, nsr_deployed
, nsr_id
, nslcmop_id
, stage
))
3103 tasks_dict_info
[task_delete_ro
] = "Removing deployment from VIM"
3105 # rest of staff will be done at finally
3107 except (ROclient
.ROClientException
, DbException
, LcmException
, N2VCException
) as e
:
3108 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3110 except asyncio
.CancelledError
:
3111 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(stage
[1]))
3112 exc
= "Operation was cancelled"
3113 except Exception as e
:
3114 exc
= traceback
.format_exc()
3115 self
.logger
.critical(logging_text
+ "Exit Exception while '{}': {}".format(stage
[1], e
), exc_info
=True)
3118 error_list
.append(str(exc
))
3120 # wait for pending tasks
3122 stage
[1] = "Waiting for terminate pending tasks."
3123 self
.logger
.debug(logging_text
+ stage
[1])
3124 error_list
+= await self
._wait
_for
_tasks
(logging_text
, tasks_dict_info
, timeout_ns_terminate
,
3126 stage
[1] = stage
[2] = ""
3127 except asyncio
.CancelledError
:
3128 error_list
.append("Cancelled")
3129 # TODO cancell all tasks
3130 except Exception as exc
:
3131 error_list
.append(str(exc
))
3132 # update status at database
3134 error_detail
= "; ".join(error_list
)
3135 # self.logger.error(logging_text + error_detail)
3136 error_description_nslcmop
= 'Stage: {}. Detail: {}'.format(stage
[0], error_detail
)
3137 error_description_nsr
= 'Operation: TERMINATING.{}, Stage {}.'.format(nslcmop_id
, stage
[0])
3139 db_nsr_update
["operational-status"] = "failed"
3140 db_nsr_update
["detailed-status"] = error_description_nsr
+ " Detail: " + error_detail
3141 db_nslcmop_update
["detailed-status"] = error_detail
3142 nslcmop_operation_state
= "FAILED"
3146 error_description_nsr
= error_description_nslcmop
= None
3147 ns_state
= "NOT_INSTANTIATED"
3148 db_nsr_update
["operational-status"] = "terminated"
3149 db_nsr_update
["detailed-status"] = "Done"
3150 db_nsr_update
["_admin.nsState"] = "NOT_INSTANTIATED"
3151 db_nslcmop_update
["detailed-status"] = "Done"
3152 nslcmop_operation_state
= "COMPLETED"
3155 self
._write
_ns
_status
(
3158 current_operation
="IDLE",
3159 current_operation_id
=None,
3160 error_description
=error_description_nsr
,
3161 error_detail
=error_detail
,
3162 other_update
=db_nsr_update
3164 self
._write
_op
_status
(
3167 error_message
=error_description_nslcmop
,
3168 operation_state
=nslcmop_operation_state
,
3169 other_update
=db_nslcmop_update
,
3171 if operation_params
:
3172 autoremove
= operation_params
.get("autoremove", False)
3173 if nslcmop_operation_state
:
3175 await self
.msg
.aiowrite("ns", "terminated", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3176 "operationState": nslcmop_operation_state
,
3177 "autoremove": autoremove
},
3179 except Exception as e
:
3180 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3182 self
.logger
.debug(logging_text
+ "Exit")
3183 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_terminate")
3185 async def _wait_for_tasks(self
, logging_text
, created_tasks_info
, timeout
, stage
, nslcmop_id
, nsr_id
=None):
3187 error_detail_list
= []
3189 pending_tasks
= list(created_tasks_info
.keys())
3190 num_tasks
= len(pending_tasks
)
3192 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3193 self
._write
_op
_status
(nslcmop_id
, stage
)
3194 while pending_tasks
:
3196 _timeout
= timeout
+ time_start
- time()
3197 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
3198 return_when
=asyncio
.FIRST_COMPLETED
)
3199 num_done
+= len(done
)
3200 if not done
: # Timeout
3201 for task
in pending_tasks
:
3202 new_error
= created_tasks_info
[task
] + ": Timeout"
3203 error_detail_list
.append(new_error
)
3204 error_list
.append(new_error
)
3207 if task
.cancelled():
3210 exc
= task
.exception()
3212 if isinstance(exc
, asyncio
.TimeoutError
):
3214 new_error
= created_tasks_info
[task
] + ": {}".format(exc
)
3215 error_list
.append(created_tasks_info
[task
])
3216 error_detail_list
.append(new_error
)
3217 if isinstance(exc
, (str, DbException
, N2VCException
, ROclient
.ROClientException
, LcmException
,
3219 self
.logger
.error(logging_text
+ new_error
)
3221 exc_traceback
= "".join(traceback
.format_exception(None, exc
, exc
.__traceback
__))
3222 self
.logger
.error(logging_text
+ created_tasks_info
[task
] + exc_traceback
)
3224 self
.logger
.debug(logging_text
+ created_tasks_info
[task
] + ": Done")
3225 stage
[1] = "{}/{}.".format(num_done
, num_tasks
)
3227 stage
[1] += " Errors: " + ". ".join(error_detail_list
) + "."
3228 if nsr_id
: # update also nsr
3229 self
.update_db_2("nsrs", nsr_id
, {"errorDescription": "Error at: " + ", ".join(error_list
),
3230 "errorDetail": ". ".join(error_detail_list
)})
3231 self
._write
_op
_status
(nslcmop_id
, stage
)
3232 return error_detail_list
3235 def _map_primitive_params(primitive_desc
, params
, instantiation_params
):
3237 Generates the params to be provided to charm before executing primitive. If user does not provide a parameter,
3238 The default-value is used. If it is between < > it look for a value at instantiation_params
3239 :param primitive_desc: portion of VNFD/NSD that describes primitive
3240 :param params: Params provided by user
3241 :param instantiation_params: Instantiation params provided by user
3242 :return: a dictionary with the calculated params
3244 calculated_params
= {}
3245 for parameter
in primitive_desc
.get("parameter", ()):
3246 param_name
= parameter
["name"]
3247 if param_name
in params
:
3248 calculated_params
[param_name
] = params
[param_name
]
3249 elif "default-value" in parameter
or "value" in parameter
:
3250 if "value" in parameter
:
3251 calculated_params
[param_name
] = parameter
["value"]
3253 calculated_params
[param_name
] = parameter
["default-value"]
3254 if isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("<") \
3255 and calculated_params
[param_name
].endswith(">"):
3256 if calculated_params
[param_name
][1:-1] in instantiation_params
:
3257 calculated_params
[param_name
] = instantiation_params
[calculated_params
[param_name
][1:-1]]
3259 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3260 format(calculated_params
[param_name
], primitive_desc
["name"]))
3262 raise LcmException("Parameter {} needed to execute primitive {} not provided".
3263 format(param_name
, primitive_desc
["name"]))
3265 if isinstance(calculated_params
[param_name
], (dict, list, tuple)):
3266 calculated_params
[param_name
] = yaml
.safe_dump(calculated_params
[param_name
], default_flow_style
=True,
3268 elif isinstance(calculated_params
[param_name
], str) and calculated_params
[param_name
].startswith("!!yaml "):
3269 calculated_params
[param_name
] = calculated_params
[param_name
][7:]
3271 # add always ns_config_info if primitive name is config
3272 if primitive_desc
["name"] == "config":
3273 if "ns_config_info" in instantiation_params
:
3274 calculated_params
["ns_config_info"] = instantiation_params
["ns_config_info"]
3275 return calculated_params
3277 def _look_for_deployed_vca(self
, deployed_vca
, member_vnf_index
, vdu_id
, vdu_count_index
, kdu_name
=None):
3278 # find vca_deployed record for this action. Raise LcmException if not found or there is not any id.
3279 for vca
in deployed_vca
:
3282 if member_vnf_index
!= vca
["member-vnf-index"] or vdu_id
!= vca
["vdu_id"]:
3284 if vdu_count_index
is not None and vdu_count_index
!= vca
["vdu_count_index"]:
3286 if kdu_name
and kdu_name
!= vca
["kdu_name"]:
3290 # vca_deployed not found
3291 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} is not "
3292 "deployed".format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3295 ee_id
= vca
.get("ee_id")
3297 raise LcmException("charm for member_vnf_index={} vdu_id={} kdu_name={} vdu_count_index={} has not "
3298 "execution environment"
3299 .format(member_vnf_index
, vdu_id
, kdu_name
, vdu_count_index
))
3302 async def _ns_execute_primitive(self
, ee_id
, primitive
, primitive_params
, retries
=0,
3303 retries_interval
=30, timeout
=None) -> (str, str):
3305 if primitive
== "config":
3306 primitive_params
= {"params": primitive_params
}
3310 output
= await asyncio
.wait_for(
3311 self
.n2vc
.exec_primitive(
3313 primitive_name
=primitive
,
3314 params_dict
=primitive_params
,
3315 progress_timeout
=self
.timeout_progress_primitive
,
3316 total_timeout
=self
.timeout_primitive
),
3317 timeout
=timeout
or self
.timeout_primitive
)
3320 except asyncio
.CancelledError
:
3322 except Exception as e
: # asyncio.TimeoutError
3323 if isinstance(e
, asyncio
.TimeoutError
):
3327 self
.logger
.debug('Error executing action {} on {} -> {}'.format(primitive
, ee_id
, e
))
3329 await asyncio
.sleep(retries_interval
, loop
=self
.loop
)
3331 return 'FAILED', str(e
)
3333 return 'COMPLETED', output
3335 except (LcmException
, asyncio
.CancelledError
):
3337 except Exception as e
:
3338 return 'FAIL', 'Error executing action {}: {}'.format(primitive
, e
)
3340 async def action(self
, nsr_id
, nslcmop_id
):
3342 # Try to lock HA task here
3343 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3344 if not task_is_locked_by_me
:
3347 logging_text
= "Task ns={} action={} ".format(nsr_id
, nslcmop_id
)
3348 self
.logger
.debug(logging_text
+ "Enter")
3349 # get all needed from database
3353 db_nslcmop_update
= {}
3354 nslcmop_operation_state
= None
3355 error_description_nslcmop
= None
3358 # wait for any previous tasks in process
3359 step
= "Waiting for previous operations to terminate"
3360 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3362 self
._write
_ns
_status
(
3365 current_operation
="RUNNING ACTION",
3366 current_operation_id
=nslcmop_id
3369 step
= "Getting information from database"
3370 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3371 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3373 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3374 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3375 vdu_id
= db_nslcmop
["operationParams"].get("vdu_id")
3376 kdu_name
= db_nslcmop
["operationParams"].get("kdu_name")
3377 vdu_count_index
= db_nslcmop
["operationParams"].get("vdu_count_index")
3378 primitive
= db_nslcmop
["operationParams"]["primitive"]
3379 primitive_params
= db_nslcmop
["operationParams"]["primitive_params"]
3380 timeout_ns_action
= db_nslcmop
["operationParams"].get("timeout_ns_action", self
.timeout_primitive
)
3383 step
= "Getting vnfr from database"
3384 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3385 step
= "Getting vnfd from database"
3386 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3388 step
= "Getting nsd from database"
3389 db_nsd
= self
.db
.get_one("nsds", {"_id": db_nsr
["nsd-id"]})
3391 # for backward compatibility
3392 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3393 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3394 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3395 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3397 # look for primitive
3398 config_primitive_desc
= None
3400 for vdu
in get_iterable(db_vnfd
, "vdu"):
3401 if vdu_id
== vdu
["id"]:
3402 for config_primitive
in deep_get(vdu
, ("vdu-configuration", "config-primitive"), ()):
3403 if config_primitive
["name"] == primitive
:
3404 config_primitive_desc
= config_primitive
3408 for kdu
in get_iterable(db_vnfd
, "kdu"):
3409 if kdu_name
== kdu
["name"]:
3410 for config_primitive
in deep_get(kdu
, ("kdu-configuration", "config-primitive"), ()):
3411 if config_primitive
["name"] == primitive
:
3412 config_primitive_desc
= config_primitive
3416 for config_primitive
in deep_get(db_vnfd
, ("vnf-configuration", "config-primitive"), ()):
3417 if config_primitive
["name"] == primitive
:
3418 config_primitive_desc
= config_primitive
3421 for config_primitive
in deep_get(db_nsd
, ("ns-configuration", "config-primitive"), ()):
3422 if config_primitive
["name"] == primitive
:
3423 config_primitive_desc
= config_primitive
3426 if not config_primitive_desc
and not (kdu_name
and primitive
in ("upgrade", "rollback", "status")):
3427 raise LcmException("Primitive {} not found at [ns|vnf|vdu]-configuration:config-primitive ".
3432 vdur
= next((x
for x
in db_vnfr
["vdur"] if x
["vdu-id-ref"] == vdu_id
), None)
3433 desc_params
= self
._format
_additional
_params
(vdur
.get("additionalParams"))
3435 kdur
= next((x
for x
in db_vnfr
["kdur"] if x
["kdu-name"] == kdu_name
), None)
3436 desc_params
= self
._format
_additional
_params
(kdur
.get("additionalParams"))
3438 desc_params
= self
._format
_additional
_params
(db_vnfr
.get("additionalParamsForVnf"))
3440 desc_params
= self
._format
_additional
_params
(db_nsr
.get("additionalParamsForNs"))
3443 kdu_action
= True if not deep_get(kdu
, ("kdu-configuration", "juju")) else False
3445 # TODO check if ns is in a proper status
3446 if kdu_name
and (primitive
in ("upgrade", "rollback", "status") or kdu_action
):
3447 # kdur and desc_params already set from before
3448 if primitive_params
:
3449 desc_params
.update(primitive_params
)
3450 # TODO Check if we will need something at vnf level
3451 for index
, kdu
in enumerate(get_iterable(nsr_deployed
, "K8s")):
3452 if kdu_name
== kdu
["kdu-name"] and kdu
["member-vnf-index"] == vnf_index
:
3455 raise LcmException("KDU '{}' for vnf '{}' not deployed".format(kdu_name
, vnf_index
))
3457 if kdu
.get("k8scluster-type") not in self
.k8scluster_map
:
3458 msg
= "unknown k8scluster-type '{}'".format(kdu
.get("k8scluster-type"))
3459 raise LcmException(msg
)
3461 db_dict
= {"collection": "nsrs",
3462 "filter": {"_id": nsr_id
},
3463 "path": "_admin.deployed.K8s.{}".format(index
)}
3464 self
.logger
.debug(logging_text
+ "Exec k8s {} on {}.{}".format(primitive
, vnf_index
, kdu_name
))
3465 step
= "Executing kdu {}".format(primitive
)
3466 if primitive
== "upgrade":
3467 if desc_params
.get("kdu_model"):
3468 kdu_model
= desc_params
.get("kdu_model")
3469 del desc_params
["kdu_model"]
3471 kdu_model
= kdu
.get("kdu-model")
3472 parts
= kdu_model
.split(sep
=":")
3474 kdu_model
= parts
[0]
3476 detailed_status
= await asyncio
.wait_for(
3477 self
.k8scluster_map
[kdu
["k8scluster-type"]].upgrade(
3478 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3479 kdu_instance
=kdu
.get("kdu-instance"),
3480 atomic
=True, kdu_model
=kdu_model
,
3481 params
=desc_params
, db_dict
=db_dict
,
3482 timeout
=timeout_ns_action
),
3483 timeout
=timeout_ns_action
+ 10)
3484 self
.logger
.debug(logging_text
+ " Upgrade of kdu {} done".format(detailed_status
))
3485 elif primitive
== "rollback":
3486 detailed_status
= await asyncio
.wait_for(
3487 self
.k8scluster_map
[kdu
["k8scluster-type"]].rollback(
3488 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3489 kdu_instance
=kdu
.get("kdu-instance"),
3491 timeout
=timeout_ns_action
)
3492 elif primitive
== "status":
3493 detailed_status
= await asyncio
.wait_for(
3494 self
.k8scluster_map
[kdu
["k8scluster-type"]].status_kdu(
3495 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3496 kdu_instance
=kdu
.get("kdu-instance")),
3497 timeout
=timeout_ns_action
)
3499 kdu_instance
= kdu
.get("kdu-instance") or "{}-{}".format(kdu
["kdu-name"], nsr_id
)
3500 params
= self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
)
3502 detailed_status
= await asyncio
.wait_for(
3503 self
.k8scluster_map
[kdu
["k8scluster-type"]].exec_primitive(
3504 cluster_uuid
=kdu
.get("k8scluster-uuid"),
3505 kdu_instance
=kdu_instance
,
3506 primitive_name
=primitive
,
3507 params
=params
, db_dict
=db_dict
,
3508 timeout
=timeout_ns_action
),
3509 timeout
=timeout_ns_action
)
3512 nslcmop_operation_state
= 'COMPLETED'
3514 detailed_status
= ''
3515 nslcmop_operation_state
= 'FAILED'
3517 nslcmop_operation_state
, detailed_status
= await self
._ns
_execute
_primitive
(
3518 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3519 member_vnf_index
=vnf_index
,
3521 vdu_count_index
=vdu_count_index
),
3522 primitive
=primitive
,
3523 primitive_params
=self
._map
_primitive
_params
(config_primitive_desc
, primitive_params
, desc_params
),
3524 timeout
=timeout_ns_action
)
3526 db_nslcmop_update
["detailed-status"] = detailed_status
3527 error_description_nslcmop
= detailed_status
if nslcmop_operation_state
== "FAILED" else ""
3528 self
.logger
.debug(logging_text
+ " task Done with result {} {}".format(nslcmop_operation_state
,
3530 return # database update is called inside finally
3532 except (DbException
, LcmException
, N2VCException
, K8sException
) as e
:
3533 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
3535 except asyncio
.CancelledError
:
3536 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
3537 exc
= "Operation was cancelled"
3538 except asyncio
.TimeoutError
:
3539 self
.logger
.error(logging_text
+ "Timeout while '{}'".format(step
))
3541 except Exception as e
:
3542 exc
= traceback
.format_exc()
3543 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
3546 db_nslcmop_update
["detailed-status"] = detailed_status
= error_description_nslcmop
= \
3547 "FAILED {}: {}".format(step
, exc
)
3548 nslcmop_operation_state
= "FAILED"
3550 self
._write
_ns
_status
(
3552 ns_state
=db_nsr
["nsState"], # TODO check if degraded. For the moment use previous status
3553 current_operation
="IDLE",
3554 current_operation_id
=None,
3555 # error_description=error_description_nsr,
3556 # error_detail=error_detail,
3557 other_update
=db_nsr_update
3560 self
._write
_op
_status
(
3563 error_message
=error_description_nslcmop
,
3564 operation_state
=nslcmop_operation_state
,
3565 other_update
=db_nslcmop_update
,
3568 if nslcmop_operation_state
:
3570 await self
.msg
.aiowrite("ns", "actioned", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
3571 "operationState": nslcmop_operation_state
},
3573 except Exception as e
:
3574 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
3575 self
.logger
.debug(logging_text
+ "Exit")
3576 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_action")
3577 return nslcmop_operation_state
, detailed_status
3579 async def scale(self
, nsr_id
, nslcmop_id
):
3581 # Try to lock HA task here
3582 task_is_locked_by_me
= self
.lcm_tasks
.lock_HA('ns', 'nslcmops', nslcmop_id
)
3583 if not task_is_locked_by_me
:
3586 logging_text
= "Task ns={} scale={} ".format(nsr_id
, nslcmop_id
)
3587 self
.logger
.debug(logging_text
+ "Enter")
3588 # get all needed from database
3591 db_nslcmop_update
= {}
3592 nslcmop_operation_state
= None
3595 # in case of error, indicates what part of scale was failed to put nsr at error status
3596 scale_process
= None
3597 old_operational_status
= ""
3598 old_config_status
= ""
3601 # wait for any previous tasks in process
3602 step
= "Waiting for previous operations to terminate"
3603 await self
.lcm_tasks
.waitfor_related_HA('ns', 'nslcmops', nslcmop_id
)
3605 self
._write
_ns
_status
(
3608 current_operation
="SCALING",
3609 current_operation_id
=nslcmop_id
3612 step
= "Getting nslcmop from database"
3613 self
.logger
.debug(step
+ " after having waited for previous tasks to be completed")
3614 db_nslcmop
= self
.db
.get_one("nslcmops", {"_id": nslcmop_id
})
3615 step
= "Getting nsr from database"
3616 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
3618 old_operational_status
= db_nsr
["operational-status"]
3619 old_config_status
= db_nsr
["config-status"]
3620 step
= "Parsing scaling parameters"
3621 # self.logger.debug(step)
3622 db_nsr_update
["operational-status"] = "scaling"
3623 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3624 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3627 nsr_deployed
= db_nsr
["_admin"].get("deployed")
3628 vnf_index
= db_nslcmop
["operationParams"].get("member_vnf_index")
3629 # vdu_id = db_nslcmop["operationParams"].get("vdu_id")
3630 # vdu_count_index = db_nslcmop["operationParams"].get("vdu_count_index")
3631 # vdu_name = db_nslcmop["operationParams"].get("vdu_name")
3634 RO_nsr_id
= nsr_deployed
["RO"]["nsr_id"]
3635 vnf_index
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]
3636 scaling_group
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]
3637 scaling_type
= db_nslcmop
["operationParams"]["scaleVnfData"]["scaleVnfType"]
3638 # scaling_policy = db_nslcmop["operationParams"]["scaleVnfData"]["scaleByStepData"].get("scaling-policy")
3640 # for backward compatibility
3641 if nsr_deployed
and isinstance(nsr_deployed
.get("VCA"), dict):
3642 nsr_deployed
["VCA"] = list(nsr_deployed
["VCA"].values())
3643 db_nsr_update
["_admin.deployed.VCA"] = nsr_deployed
["VCA"]
3644 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3646 step
= "Getting vnfr from database"
3647 db_vnfr
= self
.db
.get_one("vnfrs", {"member-vnf-index-ref": vnf_index
, "nsr-id-ref": nsr_id
})
3648 step
= "Getting vnfd from database"
3649 db_vnfd
= self
.db
.get_one("vnfds", {"_id": db_vnfr
["vnfd-id"]})
3651 step
= "Getting scaling-group-descriptor"
3652 for scaling_descriptor
in db_vnfd
["scaling-group-descriptor"]:
3653 if scaling_descriptor
["name"] == scaling_group
:
3656 raise LcmException("input parameter 'scaleByStepData':'scaling-group-descriptor':'{}' is not present "
3657 "at vnfd:scaling-group-descriptor".format(scaling_group
))
3660 # for scaling_policy_descriptor in scaling_descriptor.get("scaling-policy", ()):
3661 # cooldown_time = scaling_policy_descriptor.get("cooldown-time", 0)
3662 # if scaling_policy and scaling_policy == scaling_policy_descriptor.get("name"):
3665 # TODO check if ns is in a proper status
3666 step
= "Sending scale order to VIM"
3668 if not db_nsr
["_admin"].get("scaling-group"):
3669 self
.update_db_2("nsrs", nsr_id
, {"_admin.scaling-group": [{"name": scaling_group
, "nb-scale-op": 0}]})
3670 admin_scale_index
= 0
3672 for admin_scale_index
, admin_scale_info
in enumerate(db_nsr
["_admin"]["scaling-group"]):
3673 if admin_scale_info
["name"] == scaling_group
:
3674 nb_scale_op
= admin_scale_info
.get("nb-scale-op", 0)
3676 else: # not found, set index one plus last element and add new entry with the name
3677 admin_scale_index
+= 1
3678 db_nsr_update
["_admin.scaling-group.{}.name".format(admin_scale_index
)] = scaling_group
3679 RO_scaling_info
= []
3680 vdu_scaling_info
= {"scaling_group_name": scaling_group
, "vdu": []}
3681 if scaling_type
== "SCALE_OUT":
3682 # count if max-instance-count is reached
3683 max_instance_count
= scaling_descriptor
.get("max-instance-count", 10)
3684 # self.logger.debug("MAX_INSTANCE_COUNT is {}".format(max_instance_count))
3685 if nb_scale_op
>= max_instance_count
:
3686 raise LcmException("reached the limit of {} (max-instance-count) "
3687 "scaling-out operations for the "
3688 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3691 vdu_scaling_info
["scaling_direction"] = "OUT"
3692 vdu_scaling_info
["vdu-create"] = {}
3693 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3694 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3695 "type": "create", "count": vdu_scale_info
.get("count", 1)})
3696 vdu_scaling_info
["vdu-create"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3698 elif scaling_type
== "SCALE_IN":
3699 # count if min-instance-count is reached
3700 min_instance_count
= 0
3701 if "min-instance-count" in scaling_descriptor
and scaling_descriptor
["min-instance-count"] is not None:
3702 min_instance_count
= int(scaling_descriptor
["min-instance-count"])
3703 if nb_scale_op
<= min_instance_count
:
3704 raise LcmException("reached the limit of {} (min-instance-count) scaling-in operations for the "
3705 "scaling-group-descriptor '{}'".format(nb_scale_op
, scaling_group
))
3707 vdu_scaling_info
["scaling_direction"] = "IN"
3708 vdu_scaling_info
["vdu-delete"] = {}
3709 for vdu_scale_info
in scaling_descriptor
["vdu"]:
3710 RO_scaling_info
.append({"osm_vdu_id": vdu_scale_info
["vdu-id-ref"], "member-vnf-index": vnf_index
,
3711 "type": "delete", "count": vdu_scale_info
.get("count", 1)})
3712 vdu_scaling_info
["vdu-delete"][vdu_scale_info
["vdu-id-ref"]] = vdu_scale_info
.get("count", 1)
3714 # update VDU_SCALING_INFO with the VDUs to delete ip_addresses
3715 vdu_create
= vdu_scaling_info
.get("vdu-create")
3716 vdu_delete
= copy(vdu_scaling_info
.get("vdu-delete"))
3717 if vdu_scaling_info
["scaling_direction"] == "IN":
3718 for vdur
in reversed(db_vnfr
["vdur"]):
3719 if vdu_delete
.get(vdur
["vdu-id-ref"]):
3720 vdu_delete
[vdur
["vdu-id-ref"]] -= 1
3721 vdu_scaling_info
["vdu"].append({
3722 "name": vdur
["name"],
3723 "vdu_id": vdur
["vdu-id-ref"],
3726 for interface
in vdur
["interfaces"]:
3727 vdu_scaling_info
["vdu"][-1]["interface"].append({
3728 "name": interface
["name"],
3729 "ip_address": interface
["ip-address"],
3730 "mac_address": interface
.get("mac-address"),
3732 vdu_delete
= vdu_scaling_info
.pop("vdu-delete")
3735 step
= "Executing pre-scale vnf-config-primitive"
3736 if scaling_descriptor
.get("scaling-config-action"):
3737 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3738 if (scaling_config_action
.get("trigger") == "pre-scale-in" and scaling_type
== "SCALE_IN") \
3739 or (scaling_config_action
.get("trigger") == "pre-scale-out" and scaling_type
== "SCALE_OUT"):
3740 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3741 step
= db_nslcmop_update
["detailed-status"] = \
3742 "executing pre-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3744 # look for primitive
3745 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3746 if config_primitive
["name"] == vnf_config_primitive
:
3750 "Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:scaling-config-action"
3751 "[vnf-config-primitive-name-ref='{}'] does not match any vnf-configuration:config-"
3752 "primitive".format(scaling_group
, config_primitive
))
3754 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3755 if db_vnfr
.get("additionalParamsForVnf"):
3756 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3758 scale_process
= "VCA"
3759 db_nsr_update
["config-status"] = "configuring pre-scaling"
3760 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3762 # Pre-scale retry check: Check if this sub-operation has been executed before
3763 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3764 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'PRE-SCALE')
3765 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3766 # Skip sub-operation
3767 result
= 'COMPLETED'
3768 result_detail
= 'Done'
3769 self
.logger
.debug(logging_text
+
3770 "vnf_config_primitive={} Skipped sub-operation, result {} {}".format(
3771 vnf_config_primitive
, result
, result_detail
))
3773 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3774 # New sub-operation: Get index of this sub-operation
3775 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3776 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3777 format(vnf_config_primitive
))
3779 # retry: Get registered params for this existing sub-operation
3780 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3781 vnf_index
= op
.get('member_vnf_index')
3782 vnf_config_primitive
= op
.get('primitive')
3783 primitive_params
= op
.get('primitive_params')
3784 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
3785 format(vnf_config_primitive
))
3786 # Execute the primitive, either with new (first-time) or registered (retry) args
3787 result
, result_detail
= await self
._ns
_execute
_primitive
(
3788 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3789 member_vnf_index
=vnf_index
,
3791 vdu_count_index
=None),
3792 vnf_config_primitive
, primitive_params
)
3793 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3794 vnf_config_primitive
, result
, result_detail
))
3795 # Update operationState = COMPLETED | FAILED
3796 self
._update
_suboperation
_status
(
3797 db_nslcmop
, op_index
, result
, result_detail
)
3799 if result
== "FAILED":
3800 raise LcmException(result_detail
)
3801 db_nsr_update
["config-status"] = old_config_status
3802 scale_process
= None
3806 # Should this block be skipped if 'RO_nsr_id' == None ?
3807 # if (RO_nsr_id and RO_scaling_info):
3809 scale_process
= "RO"
3810 # Scale RO retry check: Check if this sub-operation has been executed before
3811 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3812 db_nslcmop
, vnf_index
, None, None, 'SCALE-RO', RO_nsr_id
, RO_scaling_info
)
3813 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3814 # Skip sub-operation
3815 result
= 'COMPLETED'
3816 result_detail
= 'Done'
3817 self
.logger
.debug(logging_text
+ "Skipped sub-operation RO, result {} {}".format(
3818 result
, result_detail
))
3820 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3821 # New sub-operation: Get index of this sub-operation
3822 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3823 self
.logger
.debug(logging_text
+ "New sub-operation RO")
3825 # retry: Get registered params for this existing sub-operation
3826 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3827 RO_nsr_id
= op
.get('RO_nsr_id')
3828 RO_scaling_info
= op
.get('RO_scaling_info')
3829 self
.logger
.debug(logging_text
+ "Sub-operation RO retry for primitive {}".format(
3830 vnf_config_primitive
))
3832 RO_desc
= await self
.RO
.create_action("ns", RO_nsr_id
, {"vdu-scaling": RO_scaling_info
})
3833 db_nsr_update
["_admin.scaling-group.{}.nb-scale-op".format(admin_scale_index
)] = nb_scale_op
3834 db_nsr_update
["_admin.scaling-group.{}.time".format(admin_scale_index
)] = time()
3836 RO_nslcmop_id
= RO_desc
["instance_action_id"]
3837 db_nslcmop_update
["_admin.deploy.RO"] = RO_nslcmop_id
3839 RO_task_done
= False
3840 step
= detailed_status
= "Waiting RO_task_id={} to complete the scale action.".format(RO_nslcmop_id
)
3841 detailed_status_old
= None
3842 self
.logger
.debug(logging_text
+ step
)
3844 deployment_timeout
= 1 * 3600 # One hour
3845 while deployment_timeout
> 0:
3846 if not RO_task_done
:
3847 desc
= await self
.RO
.show("ns", item_id_name
=RO_nsr_id
, extra_item
="action",
3848 extra_item_id
=RO_nslcmop_id
)
3851 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3853 ns_status
, ns_status_info
= self
.RO
.check_action_status(desc
)
3854 if ns_status
== "ERROR":
3855 raise ROclient
.ROClientException(ns_status_info
)
3856 elif ns_status
== "BUILD":
3857 detailed_status
= step
+ "; {}".format(ns_status_info
)
3858 elif ns_status
== "ACTIVE":
3860 step
= detailed_status
= "Waiting ns ready at RO. RO_id={}".format(RO_nsr_id
)
3861 self
.logger
.debug(logging_text
+ step
)
3863 assert False, "ROclient.check_action_status returns unknown {}".format(ns_status
)
3866 if ns_status
== "ERROR":
3867 raise ROclient
.ROClientException(ns_status_info
)
3868 elif ns_status
== "BUILD":
3869 detailed_status
= step
+ "; {}".format(ns_status_info
)
3870 elif ns_status
== "ACTIVE":
3871 step
= detailed_status
= \
3872 "Waiting for management IP address reported by the VIM. Updating VNFRs"
3874 self
.scale_vnfr(db_vnfr
, vdu_create
=vdu_create
, vdu_delete
=vdu_delete
)
3877 desc
= await self
.RO
.show("ns", RO_nsr_id
)
3880 self
._on
_update
_ro
_db
(nsrs_id
=nsr_id
, ro_descriptor
=desc
)
3882 # nsr_deployed["nsr_ip"] = RO.get_ns_vnf_info(desc)
3883 self
.ns_update_vnfr({db_vnfr
["member-vnf-index-ref"]: db_vnfr
}, desc
)
3885 except LcmExceptionNoMgmtIP
:
3888 assert False, "ROclient.check_ns_status returns unknown {}".format(ns_status
)
3889 if detailed_status
!= detailed_status_old
:
3890 self
._update
_suboperation
_status
(
3891 db_nslcmop
, op_index
, 'COMPLETED', detailed_status
)
3892 detailed_status_old
= db_nslcmop_update
["detailed-status"] = detailed_status
3893 self
.update_db_2("nslcmops", nslcmop_id
, db_nslcmop_update
)
3895 await asyncio
.sleep(5, loop
=self
.loop
)
3896 deployment_timeout
-= 5
3897 if deployment_timeout
<= 0:
3898 self
._update
_suboperation
_status
(
3899 db_nslcmop
, nslcmop_id
, op_index
, 'FAILED', "Timeout when waiting for ns to get ready")
3900 raise ROclient
.ROClientException("Timeout waiting ns to be ready")
3902 # update VDU_SCALING_INFO with the obtained ip_addresses
3903 if vdu_scaling_info
["scaling_direction"] == "OUT":
3904 for vdur
in reversed(db_vnfr
["vdur"]):
3905 if vdu_scaling_info
["vdu-create"].get(vdur
["vdu-id-ref"]):
3906 vdu_scaling_info
["vdu-create"][vdur
["vdu-id-ref"]] -= 1
3907 vdu_scaling_info
["vdu"].append({
3908 "name": vdur
["name"],
3909 "vdu_id": vdur
["vdu-id-ref"],
3912 for interface
in vdur
["interfaces"]:
3913 vdu_scaling_info
["vdu"][-1]["interface"].append({
3914 "name": interface
["name"],
3915 "ip_address": interface
["ip-address"],
3916 "mac_address": interface
.get("mac-address"),
3918 del vdu_scaling_info
["vdu-create"]
3920 self
._update
_suboperation
_status
(db_nslcmop
, op_index
, 'COMPLETED', 'Done')
3923 scale_process
= None
3925 self
.update_db_2("nsrs", nsr_id
, db_nsr_update
)
3928 # execute primitive service POST-SCALING
3929 step
= "Executing post-scale vnf-config-primitive"
3930 if scaling_descriptor
.get("scaling-config-action"):
3931 for scaling_config_action
in scaling_descriptor
["scaling-config-action"]:
3932 if (scaling_config_action
.get("trigger") == "post-scale-in" and scaling_type
== "SCALE_IN") \
3933 or (scaling_config_action
.get("trigger") == "post-scale-out" and scaling_type
== "SCALE_OUT"):
3934 vnf_config_primitive
= scaling_config_action
["vnf-config-primitive-name-ref"]
3935 step
= db_nslcmop_update
["detailed-status"] = \
3936 "executing post-scale scaling-config-action '{}'".format(vnf_config_primitive
)
3938 vnfr_params
= {"VDU_SCALE_INFO": vdu_scaling_info
}
3939 if db_vnfr
.get("additionalParamsForVnf"):
3940 vnfr_params
.update(db_vnfr
["additionalParamsForVnf"])
3942 # look for primitive
3943 for config_primitive
in db_vnfd
.get("vnf-configuration", {}).get("config-primitive", ()):
3944 if config_primitive
["name"] == vnf_config_primitive
:
3947 raise LcmException("Invalid vnfd descriptor at scaling-group-descriptor[name='{}']:"
3948 "scaling-config-action[vnf-config-primitive-name-ref='{}'] does not "
3949 "match any vnf-configuration:config-primitive".format(scaling_group
,
3951 scale_process
= "VCA"
3952 db_nsr_update
["config-status"] = "configuring post-scaling"
3953 primitive_params
= self
._map
_primitive
_params
(config_primitive
, {}, vnfr_params
)
3955 # Post-scale retry check: Check if this sub-operation has been executed before
3956 op_index
= self
._check
_or
_add
_scale
_suboperation
(
3957 db_nslcmop
, nslcmop_id
, vnf_index
, vnf_config_primitive
, primitive_params
, 'POST-SCALE')
3958 if op_index
== self
.SUBOPERATION_STATUS_SKIP
:
3959 # Skip sub-operation
3960 result
= 'COMPLETED'
3961 result_detail
= 'Done'
3962 self
.logger
.debug(logging_text
+
3963 "vnf_config_primitive={} Skipped sub-operation, result {} {}".
3964 format(vnf_config_primitive
, result
, result_detail
))
3966 if op_index
== self
.SUBOPERATION_STATUS_NEW
:
3967 # New sub-operation: Get index of this sub-operation
3968 op_index
= len(db_nslcmop
.get('_admin', {}).get('operations')) - 1
3969 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} New sub-operation".
3970 format(vnf_config_primitive
))
3972 # retry: Get registered params for this existing sub-operation
3973 op
= db_nslcmop
.get('_admin', {}).get('operations', [])[op_index
]
3974 vnf_index
= op
.get('member_vnf_index')
3975 vnf_config_primitive
= op
.get('primitive')
3976 primitive_params
= op
.get('primitive_params')
3977 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Sub-operation retry".
3978 format(vnf_config_primitive
))
3979 # Execute the primitive, either with new (first-time) or registered (retry) args
3980 result
, result_detail
= await self
._ns
_execute
_primitive
(
3981 self
._look
_for
_deployed
_vca
(nsr_deployed
["VCA"],
3982 member_vnf_index
=vnf_index
,
3984 vdu_count_index
=None),
3985 vnf_config_primitive
, primitive_params
)
3986 self
.logger
.debug(logging_text
+ "vnf_config_primitive={} Done with result {} {}".format(
3987 vnf_config_primitive
, result
, result_detail
))
3988 # Update operationState = COMPLETED | FAILED
3989 self
._update
_suboperation
_status
(
3990 db_nslcmop
, op_index
, result
, result_detail
)
3992 if result
== "FAILED":
3993 raise LcmException(result_detail
)
3994 db_nsr_update
["config-status"] = old_config_status
3995 scale_process
= None
3998 db_nsr_update
["detailed-status"] = "" # "scaled {} {}".format(scaling_group, scaling_type)
3999 db_nsr_update
["operational-status"] = "running" if old_operational_status
== "failed" \
4000 else old_operational_status
4001 db_nsr_update
["config-status"] = old_config_status
4003 except (ROclient
.ROClientException
, DbException
, LcmException
) as e
:
4004 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
4006 except asyncio
.CancelledError
:
4007 self
.logger
.error(logging_text
+ "Cancelled Exception while '{}'".format(step
))
4008 exc
= "Operation was cancelled"
4009 except Exception as e
:
4010 exc
= traceback
.format_exc()
4011 self
.logger
.critical(logging_text
+ "Exit Exception {} {}".format(type(e
).__name
__, e
), exc_info
=True)
4013 self
._write
_ns
_status
(
4016 current_operation
="IDLE",
4017 current_operation_id
=None
4020 db_nslcmop_update
["detailed-status"] = error_description_nslcmop
= "FAILED {}: {}".format(step
, exc
)
4021 nslcmop_operation_state
= "FAILED"
4023 db_nsr_update
["operational-status"] = old_operational_status
4024 db_nsr_update
["config-status"] = old_config_status
4025 db_nsr_update
["detailed-status"] = ""
4027 if "VCA" in scale_process
:
4028 db_nsr_update
["config-status"] = "failed"
4029 if "RO" in scale_process
:
4030 db_nsr_update
["operational-status"] = "failed"
4031 db_nsr_update
["detailed-status"] = "FAILED scaling nslcmop={} {}: {}".format(nslcmop_id
, step
,
4034 error_description_nslcmop
= None
4035 nslcmop_operation_state
= "COMPLETED"
4036 db_nslcmop_update
["detailed-status"] = "Done"
4038 self
._write
_op
_status
(
4041 error_message
=error_description_nslcmop
,
4042 operation_state
=nslcmop_operation_state
,
4043 other_update
=db_nslcmop_update
,
4046 self
._write
_ns
_status
(
4049 current_operation
="IDLE",
4050 current_operation_id
=None,
4051 other_update
=db_nsr_update
4054 if nslcmop_operation_state
:
4056 await self
.msg
.aiowrite("ns", "scaled", {"nsr_id": nsr_id
, "nslcmop_id": nslcmop_id
,
4057 "operationState": nslcmop_operation_state
},
4060 # await asyncio.sleep(cooldown_time, loop=self.loop)
4061 # await self.msg.aiowrite("ns","scaled-cooldown-time", {"nsr_id": nsr_id, "nslcmop_id": nslcmop_id})
4062 except Exception as e
:
4063 self
.logger
.error(logging_text
+ "kafka_write notification Exception {}".format(e
))
4064 self
.logger
.debug(logging_text
+ "Exit")
4065 self
.lcm_tasks
.remove("ns", nsr_id
, nslcmop_id
, "ns_scale")