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
, db
, msg
, fs
, 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
__(db
, msg
, fs
, 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 # await asyncio.sleep(15) # TODO remove. This is for test
146 self
.logger
.debug(logging_text
+ "Exit Ok VIM account created at RO_vim_account_id={}".format(desc
["uuid"]))
149 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
150 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
152 except Exception as e
:
153 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
157 db_vim_update
["_admin.operationalState"] = "ERROR"
158 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
159 # Mark the VIM 'create' HA task as erroneous
160 operation_state
= 'FAILED'
161 operation_details
= "ERROR {}: {}".format(step
, exc
)
164 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
165 # Register the VIM 'create' HA task either
166 # succesful or erroneous, or do nothing (if legacy NBI)
167 self
.lcm_tasks
.unlock_HA('vim', 'create', op_id
,
168 operationState
=operation_state
,
169 detailed_status
=operation_details
)
170 except DbException
as e
:
171 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
173 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
175 async def edit(self
, vim_content
, order_id
):
177 # HA tasks and backward compatibility:
178 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
179 # In such a case, HA is not supported by NBI, and the HA check always returns True
180 op_id
= vim_content
.pop('op_id', None)
181 if not self
.lcm_tasks
.lock_HA('vim', 'edit', op_id
):
184 vim_id
= vim_content
["_id"]
185 logging_text
= "Task vim_edit={} ".format(vim_id
)
186 self
.logger
.debug(logging_text
+ "Enter")
193 step
= "Getting vim-id='{}' from db".format(vim_id
)
195 # wait for any previous tasks in process
196 await self
.lcm_tasks
.waitfor_related_HA('vim', 'edit', op_id
)
198 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
200 if db_vim
.get("_admin") and db_vim
["_admin"].get("deployed") and db_vim
["_admin"]["deployed"].get("RO"):
201 if vim_content
.get("config") and vim_content
["config"].get("sdn-controller"):
202 step
= "Getting sdn-controller-id='{}' from db".format(vim_content
["config"]["sdn-controller"])
203 db_sdn
= self
.db
.get_one("sdns", {"_id": vim_content
["config"]["sdn-controller"]})
205 # If the VIM account has an associated SDN account, also
206 # wait for any previous tasks in process for the SDN
207 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'ANY', db_sdn
["_id"])
209 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get(
211 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
213 raise LcmException("sdn-controller={} is not available. Not deployed at RO".format(
214 vim_content
["config"]["sdn-controller"]))
216 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
217 step
= "Editing vim at RO"
218 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
219 vim_RO
= deepcopy(vim_content
)
220 vim_RO
.pop("_id", None)
221 vim_RO
.pop("_admin", None)
222 schema_version
= vim_RO
.pop("schema_version", None)
223 vim_RO
.pop("schema_type", None)
224 vim_RO
.pop("vim_tenant_name", None)
225 if "vim_type" in vim_RO
:
226 vim_RO
["type"] = vim_RO
.pop("vim_type")
227 vim_RO
.pop("vim_user", None)
228 vim_RO
.pop("vim_password", None)
230 vim_RO
["config"]["sdn-controller"] = RO_sdn_id
231 # TODO make a deep update of sdn-port-mapping
233 await RO
.edit("vim", RO_vim_id
, descriptor
=vim_RO
)
235 step
= "Editing vim-account at RO tenant"
237 if "config" in vim_content
:
238 if "sdn-controller" in vim_content
["config"]:
239 del vim_content
["config"]["sdn-controller"]
240 if "sdn-port-mapping" in vim_content
["config"]:
241 del vim_content
["config"]["sdn-port-mapping"]
242 if not vim_content
["config"]:
243 del vim_content
["config"]
244 if "vim_tenant_name" in vim_content
:
245 vim_account_RO
["vim_tenant_name"] = vim_content
["vim_tenant_name"]
246 if "vim_password" in vim_content
:
247 vim_account_RO
["vim_password"] = vim_content
["vim_password"]
248 if vim_content
.get("vim_password"):
249 vim_account_RO
["vim_password"] = self
.db
.decrypt(vim_content
["vim_password"],
250 schema_version
=schema_version
,
252 if "config" in vim_content
:
253 vim_account_RO
["config"] = vim_content
["config"]
254 if vim_content
.get("config"):
255 vim_config_encrypted_keys
= self
.vim_config_encrypted
.get(schema_version
) or \
256 self
.vim_config_encrypted
.get("default")
257 for p
in vim_config_encrypted_keys
:
258 if vim_content
["config"].get(p
):
259 vim_account_RO
["config"][p
] = self
.db
.decrypt(vim_content
["config"][p
],
260 schema_version
=schema_version
,
263 if "vim_user" in vim_content
:
264 vim_content
["vim_username"] = vim_content
["vim_user"]
265 # vim_account must be edited always even if empty in order to ensure changes are translated to RO
266 # vim_thread. RO will remove and relaunch a new thread for this vim_account
267 await RO
.edit("vim_account", RO_vim_id
, descriptor
=vim_account_RO
)
268 db_vim_update
["_admin.operationalState"] = "ENABLED"
269 # Mark the VIM 'edit' HA task as successful
270 operation_state
= 'COMPLETED'
271 operation_details
= 'Done'
273 self
.logger
.debug(logging_text
+ "Exit Ok RO_vim_id={}".format(RO_vim_id
))
276 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
277 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
279 except Exception as e
:
280 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
284 db_vim_update
["_admin.operationalState"] = "ERROR"
285 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
286 # Mark the VIM 'edit' HA task as erroneous
287 operation_state
= 'FAILED'
288 operation_details
= "ERROR {}: {}".format(step
, exc
)
291 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
292 # Register the VIM 'edit' HA task either
293 # succesful or erroneous, or do nothing (if legacy NBI)
294 self
.lcm_tasks
.unlock_HA('vim', 'edit', op_id
,
295 operationState
=operation_state
,
296 detailed_status
=operation_details
)
297 except DbException
as e
:
298 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
300 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
302 async def delete(self
, vim_content
, order_id
):
304 # HA tasks and backward compatibility:
305 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
306 # In such a case, HA is not supported by NBI, and the HA check always returns True
307 op_id
= vim_content
.pop('op_id', None)
308 if not self
.lcm_tasks
.lock_HA('vim', 'delete', op_id
):
311 vim_id
= vim_content
["_id"]
312 logging_text
= "Task vim_delete={} ".format(vim_id
)
313 self
.logger
.debug(logging_text
+ "Enter")
318 step
= "Getting vim from db"
320 # wait for any previous tasks in process
321 await self
.lcm_tasks
.waitfor_related_HA('vim', 'delete', op_id
)
323 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
324 if db_vim
.get("_admin") and db_vim
["_admin"].get("deployed") and db_vim
["_admin"]["deployed"].get("RO"):
325 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
326 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
327 step
= "Detaching vim from RO tenant"
329 await RO
.detach("vim_account", RO_vim_id
)
330 except ROclient
.ROClientException
as e
:
331 if e
.http_code
== 404: # not found
332 self
.logger
.debug(logging_text
+ "RO_vim_id={} already detached".format(RO_vim_id
))
336 step
= "Deleting vim from RO"
338 await RO
.delete("vim", RO_vim_id
)
339 except ROclient
.ROClientException
as e
:
340 if e
.http_code
== 404: # not found
341 self
.logger
.debug(logging_text
+ "RO_vim_id={} already deleted".format(RO_vim_id
))
346 self
.logger
.error(logging_text
+ "Nothing to remove at RO")
347 self
.db
.del_one("vim_accounts", {"_id": vim_id
})
349 self
.logger
.debug(logging_text
+ "Exit Ok")
352 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
353 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
355 except Exception as e
:
356 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
359 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
361 db_vim_update
["_admin.operationalState"] = "ERROR"
362 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
363 # Mark the VIM 'delete' HA task as erroneous
364 operation_state
= 'FAILED'
365 operation_details
= "ERROR {}: {}".format(step
, exc
)
366 self
.lcm_tasks
.unlock_HA('vim', 'delete', op_id
,
367 operationState
=operation_state
,
368 detailed_status
=operation_details
)
370 if db_vim
and db_vim_update
:
371 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
372 # If the VIM 'delete' HA task was succesful, the DB entry has been deleted,
373 # which means that there is nowhere to register this task, so do nothing here.
374 except DbException
as e
:
375 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
376 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
379 class WimLcm(LcmBase
):
380 # values that are encrypted at wim config because they are passwords
381 wim_config_encrypted
= ()
383 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
385 Init, Connect to database, filesystem storage, and messaging
386 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
390 self
.logger
= logging
.getLogger('lcm.vim')
392 self
.lcm_tasks
= lcm_tasks
393 self
.ro_config
= config
["ro_config"]
395 super().__init
__(db
, msg
, fs
, self
.logger
)
397 async def create(self
, wim_content
, order_id
):
399 # HA tasks and backward compatibility:
400 # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
401 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
402 # Register 'create' task here for related future HA operations
403 op_id
= wim_content
.pop('op_id', None)
404 self
.lcm_tasks
.lock_HA('wim', 'create', op_id
)
406 wim_id
= wim_content
["_id"]
407 logging_text
= "Task wim_create={} ".format(wim_id
)
408 self
.logger
.debug(logging_text
+ "Enter")
414 step
= "Getting wim-id='{}' from db".format(wim_id
)
415 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
416 db_wim_update
["_admin.deployed.RO"] = None
418 step
= "Creating wim at RO"
419 db_wim_update
["_admin.detailed-status"] = step
420 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
421 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
422 wim_RO
= deepcopy(wim_content
)
423 wim_RO
.pop("_id", None)
424 wim_RO
.pop("_admin", None)
425 schema_version
= wim_RO
.pop("schema_version", None)
426 wim_RO
.pop("schema_type", None)
427 wim_RO
.pop("wim_tenant_name", None)
428 wim_RO
["type"] = wim_RO
.pop("wim_type")
429 wim_RO
.pop("wim_user", None)
430 wim_RO
.pop("wim_password", None)
431 desc
= await RO
.create("wim", descriptor
=wim_RO
)
432 RO_wim_id
= desc
["uuid"]
433 db_wim_update
["_admin.deployed.RO"] = RO_wim_id
434 self
.logger
.debug(logging_text
+ "WIM created at RO_wim_id={}".format(RO_wim_id
))
436 step
= "Creating wim_account at RO"
437 db_wim_update
["_admin.detailed-status"] = step
438 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
440 if wim_content
.get("wim_password"):
441 wim_content
["wim_password"] = self
.db
.decrypt(wim_content
["wim_password"],
442 schema_version
=schema_version
,
444 wim_account_RO
= {"name": wim_content
["name"],
445 "user": wim_content
["user"],
446 "password": wim_content
["password"]
448 if wim_RO
.get("config"):
449 wim_account_RO
["config"] = wim_RO
["config"]
450 if "wim_port_mapping" in wim_account_RO
["config"]:
451 del wim_account_RO
["config"]["wim_port_mapping"]
452 for p
in self
.wim_config_encrypted
:
453 if wim_account_RO
["config"].get(p
):
454 wim_account_RO
["config"][p
] = self
.db
.decrypt(wim_account_RO
["config"][p
],
455 schema_version
=schema_version
,
458 desc
= await RO
.attach("wim_account", RO_wim_id
, descriptor
=wim_account_RO
)
459 db_wim_update
["_admin.deployed.RO-account"] = desc
["uuid"]
460 db_wim_update
["_admin.operationalState"] = "ENABLED"
461 db_wim_update
["_admin.detailed-status"] = "Done"
462 # Mark the WIM 'create' HA task as successful
463 operation_state
= 'COMPLETED'
464 operation_details
= 'Done'
466 self
.logger
.debug(logging_text
+ "Exit Ok WIM account created at RO_wim_account_id={}".format(desc
["uuid"]))
469 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
470 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
472 except Exception as e
:
473 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
477 db_wim_update
["_admin.operationalState"] = "ERROR"
478 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
479 # Mark the WIM 'create' HA task as erroneous
480 operation_state
= 'FAILED'
481 operation_details
= "ERROR {}: {}".format(step
, exc
)
484 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
485 # Register the WIM 'create' HA task either
486 # succesful or erroneous, or do nothing (if legacy NBI)
487 self
.lcm_tasks
.unlock_HA('wim', 'create', op_id
,
488 operationState
=operation_state
,
489 detailed_status
=operation_details
)
490 except DbException
as e
:
491 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
492 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
494 async def edit(self
, wim_content
, order_id
):
496 # HA tasks and backward compatibility:
497 # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
498 # In such a case, HA is not supported by NBI, and the HA check always returns True
499 op_id
= wim_content
.pop('op_id', None)
500 if not self
.lcm_tasks
.lock_HA('wim', 'edit', op_id
):
503 wim_id
= wim_content
["_id"]
504 logging_text
= "Task wim_edit={} ".format(wim_id
)
505 self
.logger
.debug(logging_text
+ "Enter")
511 step
= "Getting wim-id='{}' from db".format(wim_id
)
513 # wait for any previous tasks in process
514 await self
.lcm_tasks
.waitfor_related_HA('wim', 'edit', op_id
)
516 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
518 if db_wim
.get("_admin") and db_wim
["_admin"].get("deployed") and db_wim
["_admin"]["deployed"].get("RO"):
520 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO"]
521 step
= "Editing wim at RO"
522 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
523 wim_RO
= deepcopy(wim_content
)
524 wim_RO
.pop("_id", None)
525 wim_RO
.pop("_admin", None)
526 schema_version
= wim_RO
.pop("schema_version", None)
527 wim_RO
.pop("schema_type", None)
528 wim_RO
.pop("wim_tenant_name", None)
529 if "wim_type" in wim_RO
:
530 wim_RO
["type"] = wim_RO
.pop("wim_type")
531 wim_RO
.pop("wim_user", None)
532 wim_RO
.pop("wim_password", None)
533 # TODO make a deep update of wim_port_mapping
535 await RO
.edit("wim", RO_wim_id
, descriptor
=wim_RO
)
537 step
= "Editing wim-account at RO tenant"
539 if "config" in wim_content
:
540 if "wim_port_mapping" in wim_content
["config"]:
541 del wim_content
["config"]["wim_port_mapping"]
542 if not wim_content
["config"]:
543 del wim_content
["config"]
544 if "wim_tenant_name" in wim_content
:
545 wim_account_RO
["wim_tenant_name"] = wim_content
["wim_tenant_name"]
546 if "wim_password" in wim_content
:
547 wim_account_RO
["wim_password"] = wim_content
["wim_password"]
548 if wim_content
.get("wim_password"):
549 wim_account_RO
["wim_password"] = self
.db
.decrypt(wim_content
["wim_password"],
550 schema_version
=schema_version
,
552 if "config" in wim_content
:
553 wim_account_RO
["config"] = wim_content
["config"]
554 if wim_content
.get("config"):
555 for p
in self
.wim_config_encrypted
:
556 if wim_content
["config"].get(p
):
557 wim_account_RO
["config"][p
] = self
.db
.decrypt(wim_content
["config"][p
],
558 schema_version
=schema_version
,
561 if "wim_user" in wim_content
:
562 wim_content
["wim_username"] = wim_content
["wim_user"]
563 # wim_account must be edited always even if empty in order to ensure changes are translated to RO
564 # wim_thread. RO will remove and relaunch a new thread for this wim_account
565 await RO
.edit("wim_account", RO_wim_id
, descriptor
=wim_account_RO
)
566 db_wim_update
["_admin.operationalState"] = "ENABLED"
567 # Mark the WIM 'edit' HA task as successful
568 operation_state
= 'COMPLETED'
569 operation_details
= 'Done'
571 self
.logger
.debug(logging_text
+ "Exit Ok RO_wim_id={}".format(RO_wim_id
))
574 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
575 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
577 except Exception as e
:
578 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
582 db_wim_update
["_admin.operationalState"] = "ERROR"
583 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
584 # Mark the WIM 'edit' HA task as erroneous
585 operation_state
= 'FAILED'
586 operation_details
= "ERROR {}: {}".format(step
, exc
)
589 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
590 # Register the WIM 'edit' HA task either
591 # succesful or erroneous, or do nothing (if legacy NBI)
592 self
.lcm_tasks
.unlock_HA('wim', 'edit', op_id
,
593 operationState
=operation_state
,
594 detailed_status
=operation_details
)
595 except DbException
as e
:
596 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
597 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
599 async def delete(self
, wim_content
, order_id
):
601 # HA tasks and backward compatibility:
602 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
603 # In such a case, HA is not supported by NBI, and the HA check always returns True
604 op_id
= wim_content
.pop('op_id', None)
605 if not self
.lcm_tasks
.lock_HA('wim', 'delete', op_id
):
608 wim_id
= wim_content
["_id"]
609 logging_text
= "Task wim_delete={} ".format(wim_id
)
610 self
.logger
.debug(logging_text
+ "Enter")
615 step
= "Getting wim from db"
617 # wait for any previous tasks in process
618 await self
.lcm_tasks
.waitfor_related_HA('wim', 'delete', op_id
)
620 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
621 if db_wim
.get("_admin") and db_wim
["_admin"].get("deployed") and db_wim
["_admin"]["deployed"].get("RO"):
622 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO"]
623 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
624 step
= "Detaching wim from RO tenant"
626 await RO
.detach("wim_account", RO_wim_id
)
627 except ROclient
.ROClientException
as e
:
628 if e
.http_code
== 404: # not found
629 self
.logger
.debug(logging_text
+ "RO_wim_id={} already detached".format(RO_wim_id
))
633 step
= "Deleting wim from RO"
635 await RO
.delete("wim", RO_wim_id
)
636 except ROclient
.ROClientException
as e
:
637 if e
.http_code
== 404: # not found
638 self
.logger
.debug(logging_text
+ "RO_wim_id={} already deleted".format(RO_wim_id
))
643 self
.logger
.error(logging_text
+ "Nohing to remove at RO")
644 self
.db
.del_one("wim_accounts", {"_id": wim_id
})
646 self
.logger
.debug(logging_text
+ "Exit Ok")
649 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
650 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
652 except Exception as e
:
653 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
656 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
658 db_wim_update
["_admin.operationalState"] = "ERROR"
659 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
660 # Mark the WIM 'delete' HA task as erroneous
661 operation_state
= 'FAILED'
662 operation_details
= "ERROR {}: {}".format(step
, exc
)
663 self
.lcm_tasks
.unlock_HA('wim', 'delete', op_id
,
664 operationState
=operation_state
,
665 detailed_status
=operation_details
)
667 if db_wim
and db_wim_update
:
668 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
669 # If the WIM 'delete' HA task was succesful, the DB entry has been deleted,
670 # which means that there is nowhere to register this task, so do nothing here.
671 except DbException
as e
:
672 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
673 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
676 class SdnLcm(LcmBase
):
678 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
680 Init, Connect to database, filesystem storage, and messaging
681 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
685 self
.logger
= logging
.getLogger('lcm.sdn')
687 self
.lcm_tasks
= lcm_tasks
688 self
.ro_config
= config
["ro_config"]
690 super().__init
__(db
, msg
, fs
, self
.logger
)
692 async def create(self
, sdn_content
, order_id
):
694 # HA tasks and backward compatibility:
695 # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
696 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
697 # Register 'create' task here for related future HA operations
698 op_id
= sdn_content
.pop('op_id', None)
699 self
.lcm_tasks
.lock_HA('sdn', 'create', op_id
)
701 sdn_id
= sdn_content
["_id"]
702 logging_text
= "Task sdn_create={} ".format(sdn_id
)
703 self
.logger
.debug(logging_text
+ "Enter")
710 step
= "Getting sdn from db"
711 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
712 db_sdn_update
["_admin.deployed.RO"] = None
714 step
= "Creating sdn at RO"
715 db_sdn_update
["_admin.detailed-status"] = step
716 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
718 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
719 sdn_RO
= deepcopy(sdn_content
)
720 sdn_RO
.pop("_id", None)
721 sdn_RO
.pop("_admin", None)
722 schema_version
= sdn_RO
.pop("schema_version", None)
723 sdn_RO
.pop("schema_type", None)
724 sdn_RO
.pop("description", None)
725 if sdn_RO
.get("password"):
726 sdn_RO
["password"] = self
.db
.decrypt(sdn_RO
["password"], schema_version
=schema_version
, salt
=sdn_id
)
728 desc
= await RO
.create("sdn", descriptor
=sdn_RO
)
729 RO_sdn_id
= desc
["uuid"]
730 db_sdn_update
["_admin.deployed.RO"] = RO_sdn_id
731 db_sdn_update
["_admin.operationalState"] = "ENABLED"
732 self
.logger
.debug(logging_text
+ "Exit Ok RO_sdn_id={}".format(RO_sdn_id
))
733 # Mark the SDN 'create' HA task as successful
734 operation_state
= 'COMPLETED'
735 operation_details
= 'Done'
738 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
739 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
741 except Exception as e
:
742 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
746 db_sdn_update
["_admin.operationalState"] = "ERROR"
747 db_sdn_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
748 # Mark the SDN 'create' HA task as erroneous
749 operation_state
= 'FAILED'
750 operation_details
= "ERROR {}: {}".format(step
, exc
)
752 if db_sdn
and db_sdn_update
:
753 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
754 # Register the SDN 'create' HA task either
755 # succesful or erroneous, or do nothing (if legacy NBI)
756 self
.lcm_tasks
.unlock_HA('sdn', 'create', op_id
,
757 operationState
=operation_state
,
758 detailed_status
=operation_details
)
759 except DbException
as e
:
760 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
761 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
763 async def edit(self
, sdn_content
, order_id
):
765 # HA tasks and backward compatibility:
766 # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
767 # In such a case, HA is not supported by NBI, and the HA check always returns True
768 op_id
= sdn_content
.pop('op_id', None)
769 if not self
.lcm_tasks
.lock_HA('sdn', 'edit', op_id
):
772 sdn_id
= sdn_content
["_id"]
773 logging_text
= "Task sdn_edit={} ".format(sdn_id
)
774 self
.logger
.debug(logging_text
+ "Enter")
779 step
= "Getting sdn from db"
781 # wait for any previous tasks in process
782 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'edit', op_id
)
784 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
786 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
787 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
788 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
789 step
= "Editing sdn at RO"
790 sdn_RO
= deepcopy(sdn_content
)
791 sdn_RO
.pop("_id", None)
792 sdn_RO
.pop("_admin", None)
793 schema_version
= sdn_RO
.pop("schema_version", None)
794 sdn_RO
.pop("schema_type", None)
795 sdn_RO
.pop("description", None)
796 if sdn_RO
.get("password"):
797 sdn_RO
["password"] = self
.db
.decrypt(sdn_RO
["password"], schema_version
=schema_version
, salt
=sdn_id
)
799 await RO
.edit("sdn", RO_sdn_id
, descriptor
=sdn_RO
)
800 db_sdn_update
["_admin.operationalState"] = "ENABLED"
801 # Mark the SDN 'edit' HA task as successful
802 operation_state
= 'COMPLETED'
803 operation_details
= 'Done'
805 self
.logger
.debug(logging_text
+ "Exit Ok RO_sdn_id={}".format(RO_sdn_id
))
808 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
809 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
811 except Exception as e
:
812 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
816 db_sdn
["_admin.operationalState"] = "ERROR"
817 db_sdn
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
818 # Mark the SDN 'edit' HA task as erroneous
819 operation_state
= 'FAILED'
820 operation_details
= "ERROR {}: {}".format(step
, exc
)
823 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
824 # Register the SDN 'edit' HA task either
825 # succesful or erroneous, or do nothing (if legacy NBI)
826 self
.lcm_tasks
.unlock_HA('sdn', 'edit', op_id
,
827 operationState
=operation_state
,
828 detailed_status
=operation_details
)
829 except DbException
as e
:
830 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
831 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
833 async def delete(self
, sdn_content
, order_id
):
835 # HA tasks and backward compatibility:
836 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
837 # In such a case, HA is not supported by NBI, and the HA check always returns True
838 op_id
= sdn_content
.pop('op_id', None)
839 if not self
.lcm_tasks
.lock_HA('sdn', 'delete', op_id
):
842 sdn_id
= sdn_content
["_id"]
843 logging_text
= "Task sdn_delete={} ".format(sdn_id
)
844 self
.logger
.debug(logging_text
+ "Enter")
849 step
= "Getting sdn from db"
851 # wait for any previous tasks in process
852 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'delete', op_id
)
854 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
855 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
856 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
857 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
858 step
= "Deleting sdn from RO"
860 await RO
.delete("sdn", RO_sdn_id
)
861 except ROclient
.ROClientException
as e
:
862 if e
.http_code
== 404: # not found
863 self
.logger
.debug(logging_text
+ "RO_sdn_id={} already deleted".format(RO_sdn_id
))
868 self
.logger
.error(logging_text
+ "Skipping. There is not RO information at database")
869 self
.db
.del_one("sdns", {"_id": sdn_id
})
871 self
.logger
.debug("sdn_delete task sdn_id={} Exit Ok".format(sdn_id
))
874 except (ROclient
.ROClientException
, DbException
, asyncio
.CancelledError
) as e
:
875 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
877 except Exception as e
:
878 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
882 db_sdn
["_admin.operationalState"] = "ERROR"
883 db_sdn
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
884 # Mark the SDN 'delete' HA task as erroneous
885 operation_state
= 'FAILED'
886 operation_details
= "ERROR {}: {}".format(step
, exc
)
887 self
.lcm_tasks
.unlock_HA('sdn', 'delete', op_id
,
888 operationState
=operation_state
,
889 detailed_status
=operation_details
)
891 if db_sdn
and db_sdn_update
:
892 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
893 # If the SDN 'delete' HA task was succesful, the DB entry has been deleted,
894 # which means that there is nowhere to register this task, so do nothing here.
895 except DbException
as e
:
896 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
897 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
900 class K8sClusterLcm(LcmBase
):
903 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
905 Init, Connect to database, filesystem storage, and messaging
906 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
910 self
.logger
= logging
.getLogger('lcm.k8scluster')
912 self
.lcm_tasks
= lcm_tasks
913 self
.vca_config
= config
["VCA"]
917 self
.helm2_k8scluster
= K8sHelmConnector(
918 kubectl_command
=self
.vca_config
.get("kubectlpath"),
919 helm_command
=self
.vca_config
.get("helmpath"),
926 self
.helm3_k8scluster
= K8sHelm3Connector(
927 kubectl_command
=self
.vca_config
.get("kubectlpath"),
928 helm_command
=self
.vca_config
.get("helm3path"),
935 self
.juju_k8scluster
= K8sJujuConnector(
936 kubectl_command
=self
.vca_config
.get("kubectlpath"),
937 juju_command
=self
.vca_config
.get("jujupath"),
943 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 super().__init
__(db
, msg
, fs
, self
.logger
)
953 async def create(self
, k8scluster_content
, order_id
):
955 op_id
= k8scluster_content
.pop('op_id', None)
956 if not self
.lcm_tasks
.lock_HA('k8scluster', 'create', op_id
):
959 k8scluster_id
= k8scluster_content
["_id"]
960 logging_text
= "Task k8scluster_create={} ".format(k8scluster_id
)
961 self
.logger
.debug(logging_text
+ "Enter")
964 db_k8scluster_update
= {}
967 step
= "Getting k8scluster-id='{}' from db".format(k8scluster_id
)
968 self
.logger
.debug(logging_text
+ step
)
969 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": k8scluster_id
})
970 self
.db
.encrypt_decrypt_fields(db_k8scluster
.get("credentials"), 'decrypt', ['password', 'secret'],
971 schema_version
=db_k8scluster
["schema_version"], salt
=db_k8scluster
["_id"])
972 k8s_credentials
= yaml
.safe_dump(db_k8scluster
.get("credentials"))
975 init_target
= deep_get(db_k8scluster
, ("_admin", "init"))
976 step
= "Launching k8scluster init tasks"
977 for task_name
in ("helm-chart", "juju-bundle", "helm-chart-v3"):
978 if init_target
and task_name
not in init_target
:
980 task
= asyncio
.ensure_future(self
.k8s_map
[task_name
].init_env(k8s_credentials
,
981 reuse_cluster_uuid
=k8scluster_id
))
982 pending_tasks
.append(task
)
983 task2name
[task
] = task_name
987 reached_timeout
= False
991 _timeout
= max(1, self
.timeout_create
- (time() - now
)) # ensure not negative with max
992 step
= "Waiting for k8scluster init tasks"
993 done
, pending_tasks
= await asyncio
.wait(pending_tasks
, timeout
=_timeout
,
994 return_when
=asyncio
.FIRST_COMPLETED
)
996 # timeout. Set timeout is reached and process pending as if they hase been finished
999 reached_timeout
= True
1001 task_name
= task2name
[task
]
1004 elif task
.cancelled():
1007 exc
= task
.exception()
1010 error_text_list
.append("Failing init {}: {}".format(task_name
, exc
))
1011 db_k8scluster_update
["_admin.{}.error_msg".format(task_name
)] = str(exc
)
1012 db_k8scluster_update
["_admin.{}.id".format(task_name
)] = None
1013 db_k8scluster_update
["_admin.{}.operationalState".format(task_name
)] = "ERROR"
1014 self
.logger
.error(logging_text
+ "{} init fail: {}".format(task_name
, exc
),
1015 exc_info
=not isinstance(exc
, (N2VCException
, str)))
1017 k8s_id
, uninstall_sw
= task
.result()
1018 tasks_name_ok
.append(task_name
)
1019 self
.logger
.debug(logging_text
+ "{} init success. id={} created={}".format(
1020 task_name
, k8s_id
, uninstall_sw
))
1021 db_k8scluster_update
["_admin.{}.error_msg".format(task_name
)] = None
1022 db_k8scluster_update
["_admin.{}.id".format(task_name
)] = k8s_id
1023 db_k8scluster_update
["_admin.{}.created".format(task_name
)] = uninstall_sw
1024 db_k8scluster_update
["_admin.{}.operationalState".format(task_name
)] = "ENABLED"
1026 step
= "Updating database for " + task_name
1027 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1029 operation_details
= "ready for " + ", ".join(tasks_name_ok
)
1030 operation_state
= "COMPLETED"
1031 db_k8scluster_update
["_admin.operationalState"] = "ENABLED" if not error_text_list
else "DEGRADED"
1032 operation_details
+= "; " + ";".join(error_text_list
)
1034 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1035 operation_state
= "FAILED"
1036 operation_details
= ";".join(error_text_list
)
1037 db_k8scluster_update
["_admin.detailed-status"] = operation_details
1038 self
.logger
.debug(logging_text
+ "Done. Result: " + operation_state
)
1041 except Exception as e
:
1042 if isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
, asyncio
.CancelledError
)):
1043 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
1045 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1048 if exc
and db_k8scluster
:
1049 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1050 db_k8scluster_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1051 operation_state
= 'FAILED'
1052 operation_details
= "ERROR {}: {}".format(step
, exc
)
1054 if db_k8scluster
and db_k8scluster_update
:
1055 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1057 # Register the operation and unlock
1058 self
.lcm_tasks
.unlock_HA('k8scluster', 'create', op_id
,
1059 operationState
=operation_state
,
1060 detailed_status
=operation_details
)
1061 except DbException
as e
:
1062 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1063 self
.lcm_tasks
.remove("k8scluster", k8scluster_id
, order_id
)
1065 async def delete(self
, k8scluster_content
, order_id
):
1067 # HA tasks and backward compatibility:
1068 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1069 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1070 # Register 'delete' task here for related future HA operations
1071 op_id
= k8scluster_content
.pop('op_id', None)
1072 if not self
.lcm_tasks
.lock_HA('k8scluster', 'delete', op_id
):
1075 k8scluster_id
= k8scluster_content
["_id"]
1076 logging_text
= "Task k8scluster_delete={} ".format(k8scluster_id
)
1077 self
.logger
.debug(logging_text
+ "Enter")
1079 db_k8scluster
= None
1080 db_k8scluster_update
= {}
1083 step
= "Getting k8scluster='{}' from db".format(k8scluster_id
)
1084 self
.logger
.debug(logging_text
+ step
)
1085 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": k8scluster_id
})
1086 k8s_hc_id
= deep_get(db_k8scluster
, ("_admin", "helm-chart", "id"))
1087 k8s_h3c_id
= deep_get(db_k8scluster
, ("_admin", "helm-chart-v3", "id"))
1088 k8s_jb_id
= deep_get(db_k8scluster
, ("_admin", "juju-bundle", "id"))
1090 cluster_removed
= True
1091 if k8s_jb_id
: # delete in reverse order of creation
1092 step
= "Removing juju-bundle '{}'".format(k8s_jb_id
)
1093 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "juju-bundle", "created")) or False
1094 cluster_removed
= await self
.juju_k8scluster
.reset(cluster_uuid
=k8s_jb_id
, uninstall_sw
=uninstall_sw
)
1095 db_k8scluster_update
["_admin.juju-bundle.id"] = None
1096 db_k8scluster_update
["_admin.juju-bundle.operationalState"] = "DISABLED"
1099 step
= "Removing helm-chart '{}'".format(k8s_hc_id
)
1100 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "helm-chart", "created")) or False
1101 cluster_removed
= await self
.helm2_k8scluster
.reset(cluster_uuid
=k8s_hc_id
, uninstall_sw
=uninstall_sw
)
1102 db_k8scluster_update
["_admin.helm-chart.id"] = None
1103 db_k8scluster_update
["_admin.helm-chart.operationalState"] = "DISABLED"
1106 step
= "Removing helm-chart-v3 '{}'".format(k8s_hc_id
)
1107 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "helm-chart-v3", "created")) or False
1108 cluster_removed
= await self
.helm3_k8scluster
.reset(cluster_uuid
=k8s_hc_id
, uninstall_sw
=uninstall_sw
)
1109 db_k8scluster_update
["_admin.helm-chart-v3.id"] = None
1110 db_k8scluster_update
["_admin.helm-chart-v3.operationalState"] = "DISABLED"
1112 # Try to remove from cluster_inserted to clean old versions
1113 if k8s_hc_id
and cluster_removed
:
1114 step
= "Removing k8scluster='{}' from k8srepos".format(k8scluster_id
)
1115 self
.logger
.debug(logging_text
+ step
)
1116 db_k8srepo_list
= self
.db
.get_list("k8srepos", {"_admin.cluster-inserted": k8s_hc_id
})
1117 for k8srepo
in db_k8srepo_list
:
1119 cluster_list
= k8srepo
["_admin"]["cluster-inserted"]
1120 cluster_list
.remove(k8s_hc_id
)
1121 self
.update_db_2("k8srepos", k8srepo
["_id"], {"_admin.cluster-inserted": cluster_list
})
1122 except Exception as e
:
1123 self
.logger
.error("{}: {}".format(step
, e
))
1124 self
.db
.del_one("k8sclusters", {"_id": k8scluster_id
})
1125 db_k8scluster_update
= None
1126 self
.logger
.debug(logging_text
+ "Done")
1128 except Exception as e
:
1129 if isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
, asyncio
.CancelledError
)):
1130 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
1132 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1135 if exc
and db_k8scluster
:
1136 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1137 db_k8scluster_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1138 # Mark the WIM 'create' HA task as erroneous
1139 operation_state
= 'FAILED'
1140 operation_details
= "ERROR {}: {}".format(step
, exc
)
1142 operation_state
= 'COMPLETED'
1143 operation_details
= "deleted"
1146 if db_k8scluster_update
:
1147 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1148 # Register the K8scluster 'delete' HA task either
1149 # succesful or erroneous, or do nothing (if legacy NBI)
1150 self
.lcm_tasks
.unlock_HA('k8scluster', 'delete', op_id
,
1151 operationState
=operation_state
,
1152 detailed_status
=operation_details
)
1153 except DbException
as e
:
1154 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1155 self
.lcm_tasks
.remove("k8scluster", k8scluster_id
, order_id
)
1158 class K8sRepoLcm(LcmBase
):
1160 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
1162 Init, Connect to database, filesystem storage, and messaging
1163 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1167 self
.logger
= logging
.getLogger('lcm.k8srepo')
1169 self
.lcm_tasks
= lcm_tasks
1170 self
.vca_config
= config
["VCA"]
1174 super().__init
__(db
, msg
, fs
, self
.logger
)
1176 async def create(self
, k8srepo_content
, order_id
):
1178 # HA tasks and backward compatibility:
1179 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1180 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1181 # Register 'create' task here for related future HA operations
1183 op_id
= k8srepo_content
.pop('op_id', None)
1184 if not self
.lcm_tasks
.lock_HA('k8srepo', 'create', op_id
):
1187 k8srepo_id
= k8srepo_content
.get("_id")
1188 logging_text
= "Task k8srepo_create={} ".format(k8srepo_id
)
1189 self
.logger
.debug(logging_text
+ "Enter")
1192 db_k8srepo_update
= {}
1194 operation_state
= 'COMPLETED'
1195 operation_details
= ''
1197 step
= "Getting k8srepo-id='{}' from db".format(k8srepo_id
)
1198 self
.logger
.debug(logging_text
+ step
)
1199 db_k8srepo
= self
.db
.get_one("k8srepos", {"_id": k8srepo_id
})
1200 db_k8srepo_update
["_admin.operationalState"] = "ENABLED"
1201 except Exception as e
:
1202 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
),
1203 exc_info
=not isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
,
1204 asyncio
.CancelledError
)))
1207 if exc
and db_k8srepo
:
1208 db_k8srepo_update
["_admin.operationalState"] = "ERROR"
1209 db_k8srepo_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1210 # Mark the WIM 'create' HA task as erroneous
1211 operation_state
= 'FAILED'
1212 operation_details
= "ERROR {}: {}".format(step
, exc
)
1214 if db_k8srepo_update
:
1215 self
.update_db_2("k8srepos", k8srepo_id
, db_k8srepo_update
)
1216 # Register the K8srepo 'create' HA task either
1217 # succesful or erroneous, or do nothing (if legacy NBI)
1218 self
.lcm_tasks
.unlock_HA('k8srepo', 'create', op_id
,
1219 operationState
=operation_state
,
1220 detailed_status
=operation_details
)
1221 except DbException
as e
:
1222 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1223 self
.lcm_tasks
.remove("k8srepo", k8srepo_id
, order_id
)
1225 async def delete(self
, k8srepo_content
, order_id
):
1227 # HA tasks and backward compatibility:
1228 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1229 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1230 # Register 'delete' task here for related future HA operations
1231 op_id
= k8srepo_content
.pop('op_id', None)
1232 if not self
.lcm_tasks
.lock_HA('k8srepo', 'delete', op_id
):
1235 k8srepo_id
= k8srepo_content
.get("_id")
1236 logging_text
= "Task k8srepo_delete={} ".format(k8srepo_id
)
1237 self
.logger
.debug(logging_text
+ "Enter")
1240 db_k8srepo_update
= {}
1243 operation_state
= 'COMPLETED'
1244 operation_details
= ''
1246 step
= "Getting k8srepo-id='{}' from db".format(k8srepo_id
)
1247 self
.logger
.debug(logging_text
+ step
)
1248 db_k8srepo
= self
.db
.get_one("k8srepos", {"_id": k8srepo_id
})
1250 except Exception as e
:
1251 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
),
1252 exc_info
=not isinstance(e
, (LcmException
, DbException
, K8sException
, N2VCException
,
1253 asyncio
.CancelledError
)))
1256 if exc
and db_k8srepo
:
1257 db_k8srepo_update
["_admin.operationalState"] = "ERROR"
1258 db_k8srepo_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1259 # Mark the WIM 'create' HA task as erroneous
1260 operation_state
= 'FAILED'
1261 operation_details
= "ERROR {}: {}".format(step
, exc
)
1263 if db_k8srepo_update
:
1264 self
.update_db_2("k8srepos", k8srepo_id
, db_k8srepo_update
)
1265 # Register the K8srepo 'delete' HA task either
1266 # succesful or erroneous, or do nothing (if legacy NBI)
1267 self
.lcm_tasks
.unlock_HA('k8srepo', 'delete', op_id
,
1268 operationState
=operation_state
,
1269 detailed_status
=operation_details
)
1270 self
.db
.del_one("k8srepos", {"_id": k8srepo_id
})
1271 except DbException
as e
:
1272 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1273 self
.lcm_tasks
.remove("k8srepo", k8srepo_id
, order_id
)