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
23 from osm_lcm
import ROclient
24 from osm_lcm
.lcm_utils
import LcmException
, LcmBase
, deep_get
25 from n2vc
.k8s_helm_conn
import K8sHelmConnector
26 from n2vc
.k8s_helm3_conn
import K8sHelm3Connector
27 from n2vc
.k8s_juju_conn
import K8sJujuConnector
28 from n2vc
.exceptions
import K8sException
, N2VCException
29 from osm_common
.dbbase
import DbException
30 from copy
import deepcopy
33 __author__
= "Alfonso Tierno"
36 class VimLcm(LcmBase
):
37 # values that are encrypted at vim config because they are passwords
38 vim_config_encrypted
= {"1.1": ("admin_password", "nsx_password", "vcenter_password"),
39 "default": ("admin_password", "nsx_password", "vcenter_password", "vrops_password")}
41 def __init__(self
, msg
, lcm_tasks
, config
, loop
):
43 Init, Connect to database, filesystem storage, and messaging
44 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
48 self
.logger
= logging
.getLogger('lcm.vim')
50 self
.lcm_tasks
= lcm_tasks
51 self
.ro_config
= config
["ro_config"]
53 super().__init
__(msg
, self
.logger
)
55 async def create(self
, vim_content
, order_id
):
57 # HA tasks and backward compatibility:
58 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
59 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
60 # Register 'create' task here for related future HA operations
61 op_id
= vim_content
.pop('op_id', None)
62 if not self
.lcm_tasks
.lock_HA('vim', 'create', op_id
):
65 vim_id
= vim_content
["_id"]
66 logging_text
= "Task vim_create={} ".format(vim_id
)
67 self
.logger
.debug(logging_text
+ "Enter")
74 step
= "Getting vim-id='{}' from db".format(vim_id
)
75 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
76 if vim_content
.get("config") and vim_content
["config"].get("sdn-controller"):
77 step
= "Getting sdn-controller-id='{}' from db".format(vim_content
["config"]["sdn-controller"])
78 db_sdn
= self
.db
.get_one("sdns", {"_id": vim_content
["config"]["sdn-controller"]})
80 # If the VIM account has an associated SDN account, also
81 # wait for any previous tasks in process for the SDN
82 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'ANY', db_sdn
["_id"])
84 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
85 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
87 raise LcmException("sdn-controller={} is not available. Not deployed at RO".format(
88 vim_content
["config"]["sdn-controller"]))
90 step
= "Creating vim at RO"
91 db_vim_update
["_admin.deployed.RO"] = None
92 db_vim_update
["_admin.detailed-status"] = step
93 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
94 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
95 vim_RO
= deepcopy(vim_content
)
96 vim_RO
.pop("_id", None)
97 vim_RO
.pop("_admin", None)
98 schema_version
= vim_RO
.pop("schema_version", None)
99 vim_RO
.pop("schema_type", None)
100 vim_RO
.pop("vim_tenant_name", None)
101 vim_RO
["type"] = vim_RO
.pop("vim_type")
102 vim_RO
.pop("vim_user", None)
103 vim_RO
.pop("vim_password", None)
105 vim_RO
["config"]["sdn-controller"] = RO_sdn_id
106 desc
= await RO
.create("vim", descriptor
=vim_RO
)
107 RO_vim_id
= desc
["uuid"]
108 db_vim_update
["_admin.deployed.RO"] = RO_vim_id
109 self
.logger
.debug(logging_text
+ "VIM created at RO_vim_id={}".format(RO_vim_id
))
111 step
= "Creating vim_account at RO"
112 db_vim_update
["_admin.detailed-status"] = step
113 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
115 if vim_content
.get("vim_password"):
116 vim_content
["vim_password"] = self
.db
.decrypt(vim_content
["vim_password"],
117 schema_version
=schema_version
,
119 vim_account_RO
= {"vim_tenant_name": vim_content
["vim_tenant_name"],
120 "vim_username": vim_content
["vim_user"],
121 "vim_password": vim_content
["vim_password"]
123 if vim_RO
.get("config"):
124 vim_account_RO
["config"] = vim_RO
["config"]
125 if "sdn-controller" in vim_account_RO
["config"]:
126 del vim_account_RO
["config"]["sdn-controller"]
127 if "sdn-port-mapping" in vim_account_RO
["config"]:
128 del vim_account_RO
["config"]["sdn-port-mapping"]
129 vim_config_encrypted_keys
= self
.vim_config_encrypted
.get(schema_version
) or \
130 self
.vim_config_encrypted
.get("default")
131 for p
in vim_config_encrypted_keys
:
132 if vim_account_RO
["config"].get(p
):
133 vim_account_RO
["config"][p
] = self
.db
.decrypt(vim_account_RO
["config"][p
],
134 schema_version
=schema_version
,
137 desc
= await RO
.attach("vim_account", RO_vim_id
, descriptor
=vim_account_RO
)
138 db_vim_update
["_admin.deployed.RO-account"] = desc
["uuid"]
139 db_vim_update
["_admin.operationalState"] = "ENABLED"
140 db_vim_update
["_admin.detailed-status"] = "Done"
141 # Mark the VIM 'create' HA task as successful
142 operation_state
= 'COMPLETED'
143 operation_details
= 'Done'
145 self
.logger
.debug(logging_text
+ "Exit Ok VIM account created at RO_vim_account_id={}".format(desc
["uuid"]))
148 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
149 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
151 except Exception as e
:
152 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
156 db_vim_update
["_admin.operationalState"] = "ERROR"
157 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
158 # Mark the VIM 'create' HA task as erroneous
159 operation_state
= 'FAILED'
160 operation_details
= "ERROR {}: {}".format(step
, exc
)
163 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
164 # Register the VIM 'create' HA task either
165 # succesful or erroneous, or do nothing (if legacy NBI)
166 self
.lcm_tasks
.unlock_HA('vim', 'create', op_id
,
167 operationState
=operation_state
,
168 detailed_status
=operation_details
)
169 except DbException
as e
:
170 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
172 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
174 async def edit(self
, vim_content
, order_id
):
176 # HA tasks and backward compatibility:
177 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
178 # In such a case, HA is not supported by NBI, and the HA check always returns True
179 op_id
= vim_content
.pop('op_id', None)
180 if not self
.lcm_tasks
.lock_HA('vim', 'edit', op_id
):
183 vim_id
= vim_content
["_id"]
184 logging_text
= "Task vim_edit={} ".format(vim_id
)
185 self
.logger
.debug(logging_text
+ "Enter")
192 step
= "Getting vim-id='{}' from db".format(vim_id
)
194 # wait for any previous tasks in process
195 await self
.lcm_tasks
.waitfor_related_HA('vim', 'edit', op_id
)
197 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
199 if db_vim
.get("_admin") and db_vim
["_admin"].get("deployed") and db_vim
["_admin"]["deployed"].get("RO"):
200 if vim_content
.get("config") and vim_content
["config"].get("sdn-controller"):
201 step
= "Getting sdn-controller-id='{}' from db".format(vim_content
["config"]["sdn-controller"])
202 db_sdn
= self
.db
.get_one("sdns", {"_id": vim_content
["config"]["sdn-controller"]})
204 # If the VIM account has an associated SDN account, also
205 # wait for any previous tasks in process for the SDN
206 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'ANY', db_sdn
["_id"])
208 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get(
210 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
212 raise LcmException("sdn-controller={} is not available. Not deployed at RO".format(
213 vim_content
["config"]["sdn-controller"]))
215 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
216 step
= "Editing vim at RO"
217 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
218 vim_RO
= deepcopy(vim_content
)
219 vim_RO
.pop("_id", None)
220 vim_RO
.pop("_admin", None)
221 schema_version
= vim_RO
.pop("schema_version", None)
222 vim_RO
.pop("schema_type", None)
223 vim_RO
.pop("vim_tenant_name", None)
224 if "vim_type" in vim_RO
:
225 vim_RO
["type"] = vim_RO
.pop("vim_type")
226 vim_RO
.pop("vim_user", None)
227 vim_RO
.pop("vim_password", None)
229 vim_RO
["config"]["sdn-controller"] = RO_sdn_id
230 # TODO make a deep update of sdn-port-mapping
232 await RO
.edit("vim", RO_vim_id
, descriptor
=vim_RO
)
234 step
= "Editing vim-account at RO tenant"
236 if "config" in vim_content
:
237 if "sdn-controller" in vim_content
["config"]:
238 del vim_content
["config"]["sdn-controller"]
239 if "sdn-port-mapping" in vim_content
["config"]:
240 del vim_content
["config"]["sdn-port-mapping"]
241 if not vim_content
["config"]:
242 del vim_content
["config"]
243 if "vim_tenant_name" in vim_content
:
244 vim_account_RO
["vim_tenant_name"] = vim_content
["vim_tenant_name"]
245 if "vim_password" in vim_content
:
246 vim_account_RO
["vim_password"] = vim_content
["vim_password"]
247 if vim_content
.get("vim_password"):
248 vim_account_RO
["vim_password"] = self
.db
.decrypt(vim_content
["vim_password"],
249 schema_version
=schema_version
,
251 if "config" in vim_content
:
252 vim_account_RO
["config"] = vim_content
["config"]
253 if vim_content
.get("config"):
254 vim_config_encrypted_keys
= self
.vim_config_encrypted
.get(schema_version
) or \
255 self
.vim_config_encrypted
.get("default")
256 for p
in vim_config_encrypted_keys
:
257 if vim_content
["config"].get(p
):
258 vim_account_RO
["config"][p
] = self
.db
.decrypt(vim_content
["config"][p
],
259 schema_version
=schema_version
,
262 if "vim_user" in vim_content
:
263 vim_content
["vim_username"] = vim_content
["vim_user"]
264 # vim_account must be edited always even if empty in order to ensure changes are translated to RO
265 # vim_thread. RO will remove and relaunch a new thread for this vim_account
266 await RO
.edit("vim_account", RO_vim_id
, descriptor
=vim_account_RO
)
267 db_vim_update
["_admin.operationalState"] = "ENABLED"
268 # Mark the VIM 'edit' HA task as successful
269 operation_state
= 'COMPLETED'
270 operation_details
= 'Done'
272 self
.logger
.debug(logging_text
+ "Exit Ok RO_vim_id={}".format(RO_vim_id
))
275 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
276 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
278 except Exception as e
:
279 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
283 db_vim_update
["_admin.operationalState"] = "ERROR"
284 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
285 # Mark the VIM 'edit' HA task as erroneous
286 operation_state
= 'FAILED'
287 operation_details
= "ERROR {}: {}".format(step
, exc
)
290 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
291 # Register the VIM 'edit' HA task either
292 # succesful or erroneous, or do nothing (if legacy NBI)
293 self
.lcm_tasks
.unlock_HA('vim', 'edit', op_id
,
294 operationState
=operation_state
,
295 detailed_status
=operation_details
)
296 except DbException
as e
:
297 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
299 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
301 async def delete(self
, vim_content
, order_id
):
303 # HA tasks and backward compatibility:
304 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
305 # In such a case, HA is not supported by NBI, and the HA check always returns True
306 op_id
= vim_content
.pop('op_id', None)
307 if not self
.lcm_tasks
.lock_HA('vim', 'delete', op_id
):
310 vim_id
= vim_content
["_id"]
311 logging_text
= "Task vim_delete={} ".format(vim_id
)
312 self
.logger
.debug(logging_text
+ "Enter")
317 step
= "Getting vim from db"
319 # wait for any previous tasks in process
320 await self
.lcm_tasks
.waitfor_related_HA('vim', 'delete', op_id
)
321 if not self
.ro_config
.get("ng"):
322 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
323 if db_vim
.get("_admin") and db_vim
["_admin"].get("deployed") and db_vim
["_admin"]["deployed"].get("RO"):
324 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
325 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
326 step
= "Detaching vim from RO tenant"
328 await RO
.detach("vim_account", RO_vim_id
)
329 except ROclient
.ROClientException
as e
:
330 if e
.http_code
== 404: # not found
331 self
.logger
.debug(logging_text
+ "RO_vim_id={} already detached".format(RO_vim_id
))
335 step
= "Deleting vim from RO"
337 await RO
.delete("vim", RO_vim_id
)
338 except ROclient
.ROClientException
as e
:
339 if e
.http_code
== 404: # not found
340 self
.logger
.debug(logging_text
+ "RO_vim_id={} already deleted".format(RO_vim_id
))
345 self
.logger
.debug(logging_text
+ "Nothing to remove at RO")
346 self
.db
.del_one("vim_accounts", {"_id": vim_id
})
348 self
.logger
.debug(logging_text
+ "Exit Ok")
351 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
352 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
354 except Exception as e
:
355 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
358 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
360 db_vim_update
["_admin.operationalState"] = "ERROR"
361 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
362 # Mark the VIM 'delete' HA task as erroneous
363 operation_state
= 'FAILED'
364 operation_details
= "ERROR {}: {}".format(step
, exc
)
365 self
.lcm_tasks
.unlock_HA('vim', 'delete', op_id
,
366 operationState
=operation_state
,
367 detailed_status
=operation_details
)
369 if db_vim
and db_vim_update
:
370 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
371 # If the VIM 'delete' HA task was succesful, the DB entry has been deleted,
372 # which means that there is nowhere to register this task, so do nothing here.
373 except DbException
as e
:
374 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
375 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
378 class WimLcm(LcmBase
):
379 # values that are encrypted at wim config because they are passwords
380 wim_config_encrypted
= ()
382 def __init__(self
, msg
, lcm_tasks
, config
, loop
):
384 Init, Connect to database, filesystem storage, and messaging
385 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
389 self
.logger
= logging
.getLogger('lcm.vim')
391 self
.lcm_tasks
= lcm_tasks
392 self
.ro_config
= config
["ro_config"]
394 super().__init
__(msg
, self
.logger
)
396 async def create(self
, wim_content
, order_id
):
398 # HA tasks and backward compatibility:
399 # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
400 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
401 # Register 'create' task here for related future HA operations
402 op_id
= wim_content
.pop('op_id', None)
403 self
.lcm_tasks
.lock_HA('wim', 'create', op_id
)
405 wim_id
= wim_content
["_id"]
406 logging_text
= "Task wim_create={} ".format(wim_id
)
407 self
.logger
.debug(logging_text
+ "Enter")
413 step
= "Getting wim-id='{}' from db".format(wim_id
)
414 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
415 db_wim_update
["_admin.deployed.RO"] = None
417 step
= "Creating wim at RO"
418 db_wim_update
["_admin.detailed-status"] = step
419 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
420 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
421 wim_RO
= deepcopy(wim_content
)
422 wim_RO
.pop("_id", None)
423 wim_RO
.pop("_admin", None)
424 schema_version
= wim_RO
.pop("schema_version", None)
425 wim_RO
.pop("schema_type", None)
426 wim_RO
.pop("wim_tenant_name", None)
427 wim_RO
["type"] = wim_RO
.pop("wim_type")
428 wim_RO
.pop("wim_user", None)
429 wim_RO
.pop("wim_password", None)
430 desc
= await RO
.create("wim", descriptor
=wim_RO
)
431 RO_wim_id
= desc
["uuid"]
432 db_wim_update
["_admin.deployed.RO"] = RO_wim_id
433 self
.logger
.debug(logging_text
+ "WIM created at RO_wim_id={}".format(RO_wim_id
))
435 step
= "Creating wim_account at RO"
436 db_wim_update
["_admin.detailed-status"] = step
437 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
439 if wim_content
.get("wim_password"):
440 wim_content
["wim_password"] = self
.db
.decrypt(wim_content
["wim_password"],
441 schema_version
=schema_version
,
443 wim_account_RO
= {"name": wim_content
["name"],
444 "user": wim_content
["user"],
445 "password": wim_content
["password"]
447 if wim_RO
.get("config"):
448 wim_account_RO
["config"] = wim_RO
["config"]
449 if "wim_port_mapping" in wim_account_RO
["config"]:
450 del wim_account_RO
["config"]["wim_port_mapping"]
451 for p
in self
.wim_config_encrypted
:
452 if wim_account_RO
["config"].get(p
):
453 wim_account_RO
["config"][p
] = self
.db
.decrypt(wim_account_RO
["config"][p
],
454 schema_version
=schema_version
,
457 desc
= await RO
.attach("wim_account", RO_wim_id
, descriptor
=wim_account_RO
)
458 db_wim_update
["_admin.deployed.RO-account"] = desc
["uuid"]
459 db_wim_update
["_admin.operationalState"] = "ENABLED"
460 db_wim_update
["_admin.detailed-status"] = "Done"
461 # Mark the WIM 'create' HA task as successful
462 operation_state
= 'COMPLETED'
463 operation_details
= 'Done'
465 self
.logger
.debug(logging_text
+ "Exit Ok WIM account created at RO_wim_account_id={}".format(desc
["uuid"]))
468 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
469 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
471 except Exception as e
:
472 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
476 db_wim_update
["_admin.operationalState"] = "ERROR"
477 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
478 # Mark the WIM 'create' HA task as erroneous
479 operation_state
= 'FAILED'
480 operation_details
= "ERROR {}: {}".format(step
, exc
)
483 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
484 # Register the WIM 'create' HA task either
485 # succesful or erroneous, or do nothing (if legacy NBI)
486 self
.lcm_tasks
.unlock_HA('wim', 'create', op_id
,
487 operationState
=operation_state
,
488 detailed_status
=operation_details
)
489 except DbException
as e
:
490 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
491 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
493 async def edit(self
, wim_content
, order_id
):
495 # HA tasks and backward compatibility:
496 # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
497 # In such a case, HA is not supported by NBI, and the HA check always returns True
498 op_id
= wim_content
.pop('op_id', None)
499 if not self
.lcm_tasks
.lock_HA('wim', 'edit', op_id
):
502 wim_id
= wim_content
["_id"]
503 logging_text
= "Task wim_edit={} ".format(wim_id
)
504 self
.logger
.debug(logging_text
+ "Enter")
510 step
= "Getting wim-id='{}' from db".format(wim_id
)
512 # wait for any previous tasks in process
513 await self
.lcm_tasks
.waitfor_related_HA('wim', 'edit', op_id
)
515 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
517 if db_wim
.get("_admin") and db_wim
["_admin"].get("deployed") and db_wim
["_admin"]["deployed"].get("RO"):
519 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO"]
520 step
= "Editing wim at RO"
521 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
522 wim_RO
= deepcopy(wim_content
)
523 wim_RO
.pop("_id", None)
524 wim_RO
.pop("_admin", None)
525 schema_version
= wim_RO
.pop("schema_version", None)
526 wim_RO
.pop("schema_type", None)
527 wim_RO
.pop("wim_tenant_name", None)
528 if "wim_type" in wim_RO
:
529 wim_RO
["type"] = wim_RO
.pop("wim_type")
530 wim_RO
.pop("wim_user", None)
531 wim_RO
.pop("wim_password", None)
532 # TODO make a deep update of wim_port_mapping
534 await RO
.edit("wim", RO_wim_id
, descriptor
=wim_RO
)
536 step
= "Editing wim-account at RO tenant"
538 if "config" in wim_content
:
539 if "wim_port_mapping" in wim_content
["config"]:
540 del wim_content
["config"]["wim_port_mapping"]
541 if not wim_content
["config"]:
542 del wim_content
["config"]
543 if "wim_tenant_name" in wim_content
:
544 wim_account_RO
["wim_tenant_name"] = wim_content
["wim_tenant_name"]
545 if "wim_password" in wim_content
:
546 wim_account_RO
["wim_password"] = wim_content
["wim_password"]
547 if wim_content
.get("wim_password"):
548 wim_account_RO
["wim_password"] = self
.db
.decrypt(wim_content
["wim_password"],
549 schema_version
=schema_version
,
551 if "config" in wim_content
:
552 wim_account_RO
["config"] = wim_content
["config"]
553 if wim_content
.get("config"):
554 for p
in self
.wim_config_encrypted
:
555 if wim_content
["config"].get(p
):
556 wim_account_RO
["config"][p
] = self
.db
.decrypt(wim_content
["config"][p
],
557 schema_version
=schema_version
,
560 if "wim_user" in wim_content
:
561 wim_content
["wim_username"] = wim_content
["wim_user"]
562 # wim_account must be edited always even if empty in order to ensure changes are translated to RO
563 # wim_thread. RO will remove and relaunch a new thread for this wim_account
564 await RO
.edit("wim_account", RO_wim_id
, descriptor
=wim_account_RO
)
565 db_wim_update
["_admin.operationalState"] = "ENABLED"
566 # Mark the WIM 'edit' HA task as successful
567 operation_state
= 'COMPLETED'
568 operation_details
= 'Done'
570 self
.logger
.debug(logging_text
+ "Exit Ok RO_wim_id={}".format(RO_wim_id
))
573 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
574 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
576 except Exception as e
:
577 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
581 db_wim_update
["_admin.operationalState"] = "ERROR"
582 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
583 # Mark the WIM 'edit' HA task as erroneous
584 operation_state
= 'FAILED'
585 operation_details
= "ERROR {}: {}".format(step
, exc
)
588 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
589 # Register the WIM 'edit' HA task either
590 # succesful or erroneous, or do nothing (if legacy NBI)
591 self
.lcm_tasks
.unlock_HA('wim', 'edit', op_id
,
592 operationState
=operation_state
,
593 detailed_status
=operation_details
)
594 except DbException
as e
:
595 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
596 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
598 async def delete(self
, wim_content
, order_id
):
600 # HA tasks and backward compatibility:
601 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
602 # In such a case, HA is not supported by NBI, and the HA check always returns True
603 op_id
= wim_content
.pop('op_id', None)
604 if not self
.lcm_tasks
.lock_HA('wim', 'delete', op_id
):
607 wim_id
= wim_content
["_id"]
608 logging_text
= "Task wim_delete={} ".format(wim_id
)
609 self
.logger
.debug(logging_text
+ "Enter")
614 step
= "Getting wim from db"
616 # wait for any previous tasks in process
617 await self
.lcm_tasks
.waitfor_related_HA('wim', 'delete', op_id
)
619 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
620 if db_wim
.get("_admin") and db_wim
["_admin"].get("deployed") and db_wim
["_admin"]["deployed"].get("RO"):
621 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO"]
622 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
623 step
= "Detaching wim from RO tenant"
625 await RO
.detach("wim_account", RO_wim_id
)
626 except ROclient
.ROClientException
as e
:
627 if e
.http_code
== 404: # not found
628 self
.logger
.debug(logging_text
+ "RO_wim_id={} already detached".format(RO_wim_id
))
632 step
= "Deleting wim from RO"
634 await RO
.delete("wim", RO_wim_id
)
635 except ROclient
.ROClientException
as e
:
636 if e
.http_code
== 404: # not found
637 self
.logger
.debug(logging_text
+ "RO_wim_id={} already deleted".format(RO_wim_id
))
642 self
.logger
.error(logging_text
+ "Nohing to remove at RO")
643 self
.db
.del_one("wim_accounts", {"_id": wim_id
})
645 self
.logger
.debug(logging_text
+ "Exit Ok")
648 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
649 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
651 except Exception as e
:
652 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
655 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
657 db_wim_update
["_admin.operationalState"] = "ERROR"
658 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
659 # Mark the WIM 'delete' HA task as erroneous
660 operation_state
= 'FAILED'
661 operation_details
= "ERROR {}: {}".format(step
, exc
)
662 self
.lcm_tasks
.unlock_HA('wim', 'delete', op_id
,
663 operationState
=operation_state
,
664 detailed_status
=operation_details
)
666 if db_wim
and db_wim_update
:
667 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
668 # If the WIM 'delete' HA task was succesful, the DB entry has been deleted,
669 # which means that there is nowhere to register this task, so do nothing here.
670 except DbException
as e
:
671 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
672 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
675 class SdnLcm(LcmBase
):
677 def __init__(self
, msg
, lcm_tasks
, config
, loop
):
679 Init, Connect to database, filesystem storage, and messaging
680 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
684 self
.logger
= logging
.getLogger('lcm.sdn')
686 self
.lcm_tasks
= lcm_tasks
687 self
.ro_config
= config
["ro_config"]
689 super().__init
__(msg
, self
.logger
)
691 async def create(self
, sdn_content
, order_id
):
693 # HA tasks and backward compatibility:
694 # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
695 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
696 # Register 'create' task here for related future HA operations
697 op_id
= sdn_content
.pop('op_id', None)
698 self
.lcm_tasks
.lock_HA('sdn', 'create', op_id
)
700 sdn_id
= sdn_content
["_id"]
701 logging_text
= "Task sdn_create={} ".format(sdn_id
)
702 self
.logger
.debug(logging_text
+ "Enter")
709 step
= "Getting sdn from db"
710 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
711 db_sdn_update
["_admin.deployed.RO"] = None
713 step
= "Creating sdn at RO"
714 db_sdn_update
["_admin.detailed-status"] = step
715 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
717 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
718 sdn_RO
= deepcopy(sdn_content
)
719 sdn_RO
.pop("_id", None)
720 sdn_RO
.pop("_admin", None)
721 schema_version
= sdn_RO
.pop("schema_version", None)
722 sdn_RO
.pop("schema_type", None)
723 sdn_RO
.pop("description", None)
724 if sdn_RO
.get("password"):
725 sdn_RO
["password"] = self
.db
.decrypt(sdn_RO
["password"], schema_version
=schema_version
, salt
=sdn_id
)
727 desc
= await RO
.create("sdn", descriptor
=sdn_RO
)
728 RO_sdn_id
= desc
["uuid"]
729 db_sdn_update
["_admin.deployed.RO"] = RO_sdn_id
730 db_sdn_update
["_admin.operationalState"] = "ENABLED"
731 self
.logger
.debug(logging_text
+ "Exit Ok RO_sdn_id={}".format(RO_sdn_id
))
732 # Mark the SDN 'create' HA task as successful
733 operation_state
= 'COMPLETED'
734 operation_details
= 'Done'
737 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
738 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
740 except Exception as e
:
741 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
745 db_sdn_update
["_admin.operationalState"] = "ERROR"
746 db_sdn_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
747 # Mark the SDN 'create' HA task as erroneous
748 operation_state
= 'FAILED'
749 operation_details
= "ERROR {}: {}".format(step
, exc
)
751 if db_sdn
and db_sdn_update
:
752 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
753 # Register the SDN 'create' HA task either
754 # succesful or erroneous, or do nothing (if legacy NBI)
755 self
.lcm_tasks
.unlock_HA('sdn', 'create', op_id
,
756 operationState
=operation_state
,
757 detailed_status
=operation_details
)
758 except DbException
as e
:
759 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
760 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
762 async def edit(self
, sdn_content
, order_id
):
764 # HA tasks and backward compatibility:
765 # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
766 # In such a case, HA is not supported by NBI, and the HA check always returns True
767 op_id
= sdn_content
.pop('op_id', None)
768 if not self
.lcm_tasks
.lock_HA('sdn', 'edit', op_id
):
771 sdn_id
= sdn_content
["_id"]
772 logging_text
= "Task sdn_edit={} ".format(sdn_id
)
773 self
.logger
.debug(logging_text
+ "Enter")
778 step
= "Getting sdn from db"
780 # wait for any previous tasks in process
781 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'edit', op_id
)
783 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
785 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
786 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
787 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
788 step
= "Editing sdn at RO"
789 sdn_RO
= deepcopy(sdn_content
)
790 sdn_RO
.pop("_id", None)
791 sdn_RO
.pop("_admin", None)
792 schema_version
= sdn_RO
.pop("schema_version", None)
793 sdn_RO
.pop("schema_type", None)
794 sdn_RO
.pop("description", None)
795 if sdn_RO
.get("password"):
796 sdn_RO
["password"] = self
.db
.decrypt(sdn_RO
["password"], schema_version
=schema_version
, salt
=sdn_id
)
798 await RO
.edit("sdn", RO_sdn_id
, descriptor
=sdn_RO
)
799 db_sdn_update
["_admin.operationalState"] = "ENABLED"
800 # Mark the SDN 'edit' HA task as successful
801 operation_state
= 'COMPLETED'
802 operation_details
= 'Done'
804 self
.logger
.debug(logging_text
+ "Exit Ok RO_sdn_id={}".format(RO_sdn_id
))
807 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
808 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
810 except Exception as e
:
811 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
815 db_sdn
["_admin.operationalState"] = "ERROR"
816 db_sdn
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
817 # Mark the SDN 'edit' HA task as erroneous
818 operation_state
= 'FAILED'
819 operation_details
= "ERROR {}: {}".format(step
, exc
)
822 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
823 # Register the SDN 'edit' HA task either
824 # succesful or erroneous, or do nothing (if legacy NBI)
825 self
.lcm_tasks
.unlock_HA('sdn', 'edit', op_id
,
826 operationState
=operation_state
,
827 detailed_status
=operation_details
)
828 except DbException
as e
:
829 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
830 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
832 async def delete(self
, sdn_content
, order_id
):
834 # HA tasks and backward compatibility:
835 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
836 # In such a case, HA is not supported by NBI, and the HA check always returns True
837 op_id
= sdn_content
.pop('op_id', None)
838 if not self
.lcm_tasks
.lock_HA('sdn', 'delete', op_id
):
841 sdn_id
= sdn_content
["_id"]
842 logging_text
= "Task sdn_delete={} ".format(sdn_id
)
843 self
.logger
.debug(logging_text
+ "Enter")
848 step
= "Getting sdn from db"
850 # wait for any previous tasks in process
851 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'delete', op_id
)
853 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
854 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
855 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
856 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
857 step
= "Deleting sdn from RO"
859 await RO
.delete("sdn", RO_sdn_id
)
860 except ROclient
.ROClientException
as e
:
861 if e
.http_code
== 404: # not found
862 self
.logger
.debug(logging_text
+ "RO_sdn_id={} already deleted".format(RO_sdn_id
))
867 self
.logger
.error(logging_text
+ "Skipping. There is not RO information at database")
868 self
.db
.del_one("sdns", {"_id": sdn_id
})
870 self
.logger
.debug("sdn_delete task sdn_id={} Exit Ok".format(sdn_id
))
873 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
874 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
876 except Exception as e
:
877 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
881 db_sdn
["_admin.operationalState"] = "ERROR"
882 db_sdn
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
883 # Mark the SDN 'delete' HA task as erroneous
884 operation_state
= 'FAILED'
885 operation_details
= "ERROR {}: {}".format(step
, exc
)
886 self
.lcm_tasks
.unlock_HA('sdn', 'delete', op_id
,
887 operationState
=operation_state
,
888 detailed_status
=operation_details
)
890 if db_sdn
and db_sdn_update
:
891 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
892 # If the SDN 'delete' HA task was succesful, the DB entry has been deleted,
893 # which means that there is nowhere to register this task, so do nothing here.
894 except DbException
as e
:
895 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
896 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
899 class K8sClusterLcm(LcmBase
):
902 def __init__(self
, msg
, lcm_tasks
, config
, loop
):
904 Init, Connect to database, filesystem storage, and messaging
905 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
909 self
.logger
= logging
.getLogger('lcm.k8scluster')
911 self
.lcm_tasks
= lcm_tasks
912 self
.vca_config
= config
["VCA"]
914 super().__init
__(msg
, self
.logger
)
916 self
.helm2_k8scluster
= K8sHelmConnector(
917 kubectl_command
=self
.vca_config
.get("kubectlpath"),
918 helm_command
=self
.vca_config
.get("helmpath"),
925 self
.helm3_k8scluster
= K8sHelm3Connector(
926 kubectl_command
=self
.vca_config
.get("kubectlpath"),
927 helm_command
=self
.vca_config
.get("helm3path"),
934 self
.juju_k8scluster
= K8sJujuConnector(
935 kubectl_command
=self
.vca_config
.get("kubectlpath"),
936 juju_command
=self
.vca_config
.get("jujupath"),
940 vca_config
=self
.vca_config
,
946 "helm-chart": self
.helm2_k8scluster
,
947 "helm-chart-v3": self
.helm3_k8scluster
,
948 "juju-bundle": self
.juju_k8scluster
,
951 async def create(self
, k8scluster_content
, order_id
):
953 op_id
= k8scluster_content
.pop('op_id', None)
954 if not self
.lcm_tasks
.lock_HA('k8scluster', 'create', op_id
):
957 k8scluster_id
= k8scluster_content
["_id"]
958 logging_text
= "Task k8scluster_create={} ".format(k8scluster_id
)
959 self
.logger
.debug(logging_text
+ "Enter")
962 db_k8scluster_update
= {}
965 step
= "Getting k8scluster-id='{}' from db".format(k8scluster_id
)
966 self
.logger
.debug(logging_text
+ step
)
967 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": k8scluster_id
})
968 self
.db
.encrypt_decrypt_fields(db_k8scluster
.get("credentials"), 'decrypt', ['password', 'secret'],
969 schema_version
=db_k8scluster
["schema_version"], salt
=db_k8scluster
["_id"])
970 k8s_credentials
= yaml
.safe_dump(db_k8scluster
.get("credentials"))
973 init_target
= deep_get(db_k8scluster
, ("_admin", "init"))
974 step
= "Launching k8scluster init tasks"
975 for task_name
in ("helm-chart", "juju-bundle", "helm-chart-v3"):
976 if init_target
and task_name
not in init_target
:
978 task
= asyncio
.ensure_future(self
.k8s_map
[task_name
].init_env(k8s_credentials
,
979 reuse_cluster_uuid
=k8scluster_id
))
980 pending_tasks
.append(task
)
981 task2name
[task
] = task_name
985 reached_timeout
= False
989 _timeout
= max(1, self
.timeout_create
- (time() - now
)) # ensure not negative with max
990 step
= "Waiting for k8scluster init tasks"
991 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
992 return_when
=asyncio
.FIRST_COMPLETED
)
994 # timeout. Set timeout is reached and process pending as if they hase been finished
997 reached_timeout
= True
999 task_name
= task2name
[task
]
1002 elif task
.cancelled():
1005 exc
= task
.exception()
1008 error_text_list
.append("Failing init {}: {}".format(task_name
, exc
))
1009 db_k8scluster_update
["_admin.{}.error_msg".format(task_name
)] = str(exc
)
1010 db_k8scluster_update
["_admin.{}.id".format(task_name
)] = None
1011 db_k8scluster_update
["_admin.{}.operationalState".format(task_name
)] = "ERROR"
1012 self
.logger
.error(logging_text
+ "{} init fail: {}".format(task_name
, exc
),
1013 exc_info
=not isinstance(exc
, (N2VCException
, str)))
1015 k8s_id
, uninstall_sw
= task
.result()
1016 tasks_name_ok
.append(task_name
)
1017 self
.logger
.debug(logging_text
+ "{} init success. id={} created={}".format(
1018 task_name
, k8s_id
, uninstall_sw
))
1019 db_k8scluster_update
["_admin.{}.error_msg".format(task_name
)] = None
1020 db_k8scluster_update
["_admin.{}.id".format(task_name
)] = k8s_id
1021 db_k8scluster_update
["_admin.{}.created".format(task_name
)] = uninstall_sw
1022 db_k8scluster_update
["_admin.{}.operationalState".format(task_name
)] = "ENABLED"
1024 step
= "Updating database for " + task_name
1025 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1027 operation_details
= "ready for " + ", ".join(tasks_name_ok
)
1028 operation_state
= "COMPLETED"
1029 db_k8scluster_update
["_admin.operationalState"] = "ENABLED" if not error_text_list
else "DEGRADED"
1030 operation_details
+= "; " + ";".join(error_text_list
)
1032 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1033 operation_state
= "FAILED"
1034 operation_details
= ";".join(error_text_list
)
1035 db_k8scluster_update
["_admin.detailed-status"] = operation_details
1036 self
.logger
.debug(logging_text
+ "Done. Result: " + operation_state
)
1039 except Exception as e
:
1040 if isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
, asyncio
.CancelledError
)):
1041 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
1043 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1046 if exc
and db_k8scluster
:
1047 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1048 db_k8scluster_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1049 operation_state
= 'FAILED'
1050 operation_details
= "ERROR {}: {}".format(step
, exc
)
1052 if db_k8scluster
and db_k8scluster_update
:
1053 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1055 # Register the operation and unlock
1056 self
.lcm_tasks
.unlock_HA('k8scluster', 'create', op_id
,
1057 operationState
=operation_state
,
1058 detailed_status
=operation_details
)
1059 except DbException
as e
:
1060 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1061 self
.lcm_tasks
.remove("k8scluster", k8scluster_id
, order_id
)
1063 async def delete(self
, k8scluster_content
, order_id
):
1065 # HA tasks and backward compatibility:
1066 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1067 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1068 # Register 'delete' task here for related future HA operations
1069 op_id
= k8scluster_content
.pop('op_id', None)
1070 if not self
.lcm_tasks
.lock_HA('k8scluster', 'delete', op_id
):
1073 k8scluster_id
= k8scluster_content
["_id"]
1074 logging_text
= "Task k8scluster_delete={} ".format(k8scluster_id
)
1075 self
.logger
.debug(logging_text
+ "Enter")
1077 db_k8scluster
= None
1078 db_k8scluster_update
= {}
1081 step
= "Getting k8scluster='{}' from db".format(k8scluster_id
)
1082 self
.logger
.debug(logging_text
+ step
)
1083 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": k8scluster_id
})
1084 k8s_hc_id
= deep_get(db_k8scluster
, ("_admin", "helm-chart", "id"))
1085 k8s_h3c_id
= deep_get(db_k8scluster
, ("_admin", "helm-chart-v3", "id"))
1086 k8s_jb_id
= deep_get(db_k8scluster
, ("_admin", "juju-bundle", "id"))
1088 cluster_removed
= True
1089 if k8s_jb_id
: # delete in reverse order of creation
1090 step
= "Removing juju-bundle '{}'".format(k8s_jb_id
)
1091 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "juju-bundle", "created")) or False
1092 cluster_removed
= await self
.juju_k8scluster
.reset(cluster_uuid
=k8s_jb_id
, uninstall_sw
=uninstall_sw
)
1093 db_k8scluster_update
["_admin.juju-bundle.id"] = None
1094 db_k8scluster_update
["_admin.juju-bundle.operationalState"] = "DISABLED"
1097 step
= "Removing helm-chart '{}'".format(k8s_hc_id
)
1098 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "helm-chart", "created")) or False
1099 cluster_removed
= await self
.helm2_k8scluster
.reset(cluster_uuid
=k8s_hc_id
, uninstall_sw
=uninstall_sw
)
1100 db_k8scluster_update
["_admin.helm-chart.id"] = None
1101 db_k8scluster_update
["_admin.helm-chart.operationalState"] = "DISABLED"
1104 step
= "Removing helm-chart-v3 '{}'".format(k8s_hc_id
)
1105 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "helm-chart-v3", "created")) or False
1106 cluster_removed
= await self
.helm3_k8scluster
.reset(cluster_uuid
=k8s_h3c_id
, uninstall_sw
=uninstall_sw
)
1107 db_k8scluster_update
["_admin.helm-chart-v3.id"] = None
1108 db_k8scluster_update
["_admin.helm-chart-v3.operationalState"] = "DISABLED"
1110 # Try to remove from cluster_inserted to clean old versions
1111 if k8s_hc_id
and cluster_removed
:
1112 step
= "Removing k8scluster='{}' from k8srepos".format(k8scluster_id
)
1113 self
.logger
.debug(logging_text
+ step
)
1114 db_k8srepo_list
= self
.db
.get_list("k8srepos", {"_admin.cluster-inserted": k8s_hc_id
})
1115 for k8srepo
in db_k8srepo_list
:
1117 cluster_list
= k8srepo
["_admin"]["cluster-inserted"]
1118 cluster_list
.remove(k8s_hc_id
)
1119 self
.update_db_2("k8srepos", k8srepo
["_id"], {"_admin.cluster-inserted": cluster_list
})
1120 except Exception as e
:
1121 self
.logger
.error("{}: {}".format(step
, e
))
1122 self
.db
.del_one("k8sclusters", {"_id": k8scluster_id
})
1123 db_k8scluster_update
= None
1124 self
.logger
.debug(logging_text
+ "Done")
1126 except Exception as e
:
1127 if isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
, asyncio
.CancelledError
)):
1128 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
1130 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1133 if exc
and db_k8scluster
:
1134 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1135 db_k8scluster_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1136 # Mark the WIM 'create' HA task as erroneous
1137 operation_state
= 'FAILED'
1138 operation_details
= "ERROR {}: {}".format(step
, exc
)
1140 operation_state
= 'COMPLETED'
1141 operation_details
= "deleted"
1144 if db_k8scluster_update
:
1145 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1146 # Register the K8scluster 'delete' HA task either
1147 # succesful or erroneous, or do nothing (if legacy NBI)
1148 self
.lcm_tasks
.unlock_HA('k8scluster', 'delete', op_id
,
1149 operationState
=operation_state
,
1150 detailed_status
=operation_details
)
1151 except DbException
as e
:
1152 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1153 self
.lcm_tasks
.remove("k8scluster", k8scluster_id
, order_id
)
1156 class K8sRepoLcm(LcmBase
):
1158 def __init__(self
, msg
, lcm_tasks
, config
, loop
):
1160 Init, Connect to database, filesystem storage, and messaging
1161 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1165 self
.logger
= logging
.getLogger('lcm.k8srepo')
1167 self
.lcm_tasks
= lcm_tasks
1168 self
.vca_config
= config
["VCA"]
1170 super().__init
__(msg
, self
.logger
)
1172 self
.k8srepo
= K8sHelmConnector(
1173 kubectl_command
=self
.vca_config
.get("kubectlpath"),
1174 helm_command
=self
.vca_config
.get("helmpath"),
1181 async def create(self
, k8srepo_content
, order_id
):
1183 # HA tasks and backward compatibility:
1184 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1185 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1186 # Register 'create' task here for related future HA operations
1188 op_id
= k8srepo_content
.pop('op_id', None)
1189 if not self
.lcm_tasks
.lock_HA('k8srepo', 'create', op_id
):
1192 k8srepo_id
= k8srepo_content
.get("_id")
1193 logging_text
= "Task k8srepo_create={} ".format(k8srepo_id
)
1194 self
.logger
.debug(logging_text
+ "Enter")
1197 db_k8srepo_update
= {}
1199 operation_state
= 'COMPLETED'
1200 operation_details
= ''
1202 step
= "Getting k8srepo-id='{}' from db".format(k8srepo_id
)
1203 self
.logger
.debug(logging_text
+ step
)
1204 db_k8srepo
= self
.db
.get_one("k8srepos", {"_id": k8srepo_id
})
1205 db_k8srepo_update
["_admin.operationalState"] = "ENABLED"
1206 except Exception as e
:
1207 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
),
1208 exc_info
=not isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
,
1209 asyncio
.CancelledError
)))
1212 if exc
and db_k8srepo
:
1213 db_k8srepo_update
["_admin.operationalState"] = "ERROR"
1214 db_k8srepo_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1215 # Mark the WIM 'create' HA task as erroneous
1216 operation_state
= 'FAILED'
1217 operation_details
= "ERROR {}: {}".format(step
, exc
)
1219 if db_k8srepo_update
:
1220 self
.update_db_2("k8srepos", k8srepo_id
, db_k8srepo_update
)
1221 # Register the K8srepo 'create' HA task either
1222 # succesful or erroneous, or do nothing (if legacy NBI)
1223 self
.lcm_tasks
.unlock_HA('k8srepo', 'create', op_id
,
1224 operationState
=operation_state
,
1225 detailed_status
=operation_details
)
1226 except DbException
as e
:
1227 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1228 self
.lcm_tasks
.remove("k8srepo", k8srepo_id
, order_id
)
1230 async def delete(self
, k8srepo_content
, order_id
):
1232 # HA tasks and backward compatibility:
1233 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1234 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1235 # Register 'delete' task here for related future HA operations
1236 op_id
= k8srepo_content
.pop('op_id', None)
1237 if not self
.lcm_tasks
.lock_HA('k8srepo', 'delete', op_id
):
1240 k8srepo_id
= k8srepo_content
.get("_id")
1241 logging_text
= "Task k8srepo_delete={} ".format(k8srepo_id
)
1242 self
.logger
.debug(logging_text
+ "Enter")
1245 db_k8srepo_update
= {}
1248 operation_state
= 'COMPLETED'
1249 operation_details
= ''
1251 step
= "Getting k8srepo-id='{}' from db".format(k8srepo_id
)
1252 self
.logger
.debug(logging_text
+ step
)
1253 db_k8srepo
= self
.db
.get_one("k8srepos", {"_id": k8srepo_id
})
1255 except Exception as e
:
1256 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
),
1257 exc_info
=not isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
,
1258 asyncio
.CancelledError
)))
1261 if exc
and db_k8srepo
:
1262 db_k8srepo_update
["_admin.operationalState"] = "ERROR"
1263 db_k8srepo_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1264 # Mark the WIM 'create' HA task as erroneous
1265 operation_state
= 'FAILED'
1266 operation_details
= "ERROR {}: {}".format(step
, exc
)
1268 if db_k8srepo_update
:
1269 self
.update_db_2("k8srepos", k8srepo_id
, db_k8srepo_update
)
1270 # Register the K8srepo 'delete' HA task either
1271 # succesful or erroneous, or do nothing (if legacy NBI)
1272 self
.lcm_tasks
.unlock_HA('k8srepo', 'delete', op_id
,
1273 operationState
=operation_state
,
1274 detailed_status
=operation_details
)
1275 self
.db
.del_one("k8srepos", {"_id": k8srepo_id
})
1276 except DbException
as e
:
1277 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1278 self
.lcm_tasks
.remove("k8srepo", k8srepo_id
, order_id
)