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
21 import logging
.handlers
22 from osm_lcm
import ROclient
23 from osm_lcm
.lcm_utils
import LcmException
, LcmBase
, deep_get
24 from n2vc
.k8s_helm_conn
import K8sHelmConnector
25 from n2vc
.k8s_juju_conn
import K8sJujuConnector
26 from n2vc
.exceptions
import K8sException
, N2VCException
27 from osm_common
.dbbase
import DbException
28 from copy
import deepcopy
30 __author__
= "Alfonso Tierno"
33 class VimLcm(LcmBase
):
34 # values that are encrypted at vim config because they are passwords
35 vim_config_encrypted
= {"1.1": ("admin_password", "nsx_password", "vcenter_password"),
36 "default": ("admin_password", "nsx_password", "vcenter_password", "vrops_password")}
38 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
40 Init, Connect to database, filesystem storage, and messaging
41 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
45 self
.logger
= logging
.getLogger('lcm.vim')
47 self
.lcm_tasks
= lcm_tasks
48 self
.ro_config
= config
["ro_config"]
50 super().__init
__(db
, msg
, fs
, self
.logger
)
52 async def create(self
, vim_content
, order_id
):
54 # HA tasks and backward compatibility:
55 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
56 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
57 # Register 'create' task here for related future HA operations
58 op_id
= vim_content
.pop('op_id', None)
59 if not self
.lcm_tasks
.lock_HA('vim', 'create', op_id
):
62 vim_id
= vim_content
["_id"]
63 vim_content
.pop("op_id", None)
64 logging_text
= "Task vim_create={} ".format(vim_id
)
65 self
.logger
.debug(logging_text
+ "Enter")
71 operationState_HA
= ''
72 detailed_status_HA
= ''
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 operationState_HA
= 'COMPLETED'
143 detailed_status_HA
= '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
) 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 operationState_HA
= 'FAILED'
161 detailed_status_HA
= "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
.register_HA('vim', 'create', op_id
,
168 operationState
=operationState_HA
,
169 detailed_status
=detailed_status_HA
)
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 vim_content
.pop("op_id", None)
186 logging_text
= "Task vim_edit={} ".format(vim_id
)
187 self
.logger
.debug(logging_text
+ "Enter")
194 operationState_HA
= ''
195 detailed_status_HA
= ''
196 step
= "Getting vim-id='{}' from db".format(vim_id
)
198 # wait for any previous tasks in process
199 await self
.lcm_tasks
.waitfor_related_HA('vim', 'edit', op_id
)
201 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
203 if db_vim
.get("_admin") and db_vim
["_admin"].get("deployed") and db_vim
["_admin"]["deployed"].get("RO"):
204 if vim_content
.get("config") and vim_content
["config"].get("sdn-controller"):
205 step
= "Getting sdn-controller-id='{}' from db".format(vim_content
["config"]["sdn-controller"])
206 db_sdn
= self
.db
.get_one("sdns", {"_id": vim_content
["config"]["sdn-controller"]})
208 # If the VIM account has an associated SDN account, also
209 # wait for any previous tasks in process for the SDN
210 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'ANY', db_sdn
["_id"])
212 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get(
214 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
216 raise LcmException("sdn-controller={} is not available. Not deployed at RO".format(
217 vim_content
["config"]["sdn-controller"]))
219 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
220 step
= "Editing vim at RO"
221 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
222 vim_RO
= deepcopy(vim_content
)
223 vim_RO
.pop("_id", None)
224 vim_RO
.pop("_admin", None)
225 schema_version
= vim_RO
.pop("schema_version", None)
226 vim_RO
.pop("schema_type", None)
227 vim_RO
.pop("vim_tenant_name", None)
228 if "vim_type" in vim_RO
:
229 vim_RO
["type"] = vim_RO
.pop("vim_type")
230 vim_RO
.pop("vim_user", None)
231 vim_RO
.pop("vim_password", None)
233 vim_RO
["config"]["sdn-controller"] = RO_sdn_id
234 # TODO make a deep update of sdn-port-mapping
236 await RO
.edit("vim", RO_vim_id
, descriptor
=vim_RO
)
238 step
= "Editing vim-account at RO tenant"
240 if "config" in vim_content
:
241 if "sdn-controller" in vim_content
["config"]:
242 del vim_content
["config"]["sdn-controller"]
243 if "sdn-port-mapping" in vim_content
["config"]:
244 del vim_content
["config"]["sdn-port-mapping"]
245 if not vim_content
["config"]:
246 del vim_content
["config"]
247 if "vim_tenant_name" in vim_content
:
248 vim_account_RO
["vim_tenant_name"] = vim_content
["vim_tenant_name"]
249 if "vim_password" in vim_content
:
250 vim_account_RO
["vim_password"] = vim_content
["vim_password"]
251 if vim_content
.get("vim_password"):
252 vim_account_RO
["vim_password"] = self
.db
.decrypt(vim_content
["vim_password"],
253 schema_version
=schema_version
,
255 if "config" in vim_content
:
256 vim_account_RO
["config"] = vim_content
["config"]
257 if vim_content
.get("config"):
258 vim_config_encrypted_keys
= self
.vim_config_encrypted
.get(schema_version
) or \
259 self
.vim_config_encrypted
.get("default")
260 for p
in vim_config_encrypted_keys
:
261 if vim_content
["config"].get(p
):
262 vim_account_RO
["config"][p
] = self
.db
.decrypt(vim_content
["config"][p
],
263 schema_version
=schema_version
,
266 if "vim_user" in vim_content
:
267 vim_content
["vim_username"] = vim_content
["vim_user"]
268 # vim_account must be edited always even if empty in order to ensure changes are translated to RO
269 # vim_thread. RO will remove and relaunch a new thread for this vim_account
270 await RO
.edit("vim_account", RO_vim_id
, descriptor
=vim_account_RO
)
271 db_vim_update
["_admin.operationalState"] = "ENABLED"
272 # Mark the VIM 'edit' HA task as successful
273 operationState_HA
= 'COMPLETED'
274 detailed_status_HA
= 'Done'
276 self
.logger
.debug(logging_text
+ "Exit Ok RO_vim_id={}".format(RO_vim_id
))
279 except (ROclient
.ROClientException
, DbException
) as e
:
280 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
282 except Exception as e
:
283 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
287 db_vim_update
["_admin.operationalState"] = "ERROR"
288 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
289 # Mark the VIM 'edit' HA task as erroneous
290 operationState_HA
= 'FAILED'
291 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
294 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
295 # Register the VIM 'edit' HA task either
296 # succesful or erroneous, or do nothing (if legacy NBI)
297 self
.lcm_tasks
.register_HA('vim', 'edit', op_id
,
298 operationState
=operationState_HA
,
299 detailed_status
=detailed_status_HA
)
300 except DbException
as e
:
301 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
303 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
305 async def delete(self
, vim_content
, order_id
):
307 # HA tasks and backward compatibility:
308 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
309 # In such a case, HA is not supported by NBI, and the HA check always returns True
310 op_id
= vim_content
.pop('op_id', None)
311 if not self
.lcm_tasks
.lock_HA('vim', 'delete', op_id
):
314 vim_id
= vim_content
["_id"]
315 logging_text
= "Task vim_delete={} ".format(vim_id
)
316 self
.logger
.debug(logging_text
+ "Enter")
321 operationState_HA
= ''
322 detailed_status_HA
= ''
323 step
= "Getting vim from db"
325 # wait for any previous tasks in process
326 await self
.lcm_tasks
.waitfor_related_HA('vim', 'delete', op_id
)
328 db_vim
= self
.db
.get_one("vim_accounts", {"_id": vim_id
})
329 if db_vim
.get("_admin") and db_vim
["_admin"].get("deployed") and db_vim
["_admin"]["deployed"].get("RO"):
330 RO_vim_id
= db_vim
["_admin"]["deployed"]["RO"]
331 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
332 step
= "Detaching vim from RO tenant"
334 await RO
.detach("vim_account", RO_vim_id
)
335 except ROclient
.ROClientException
as e
:
336 if e
.http_code
== 404: # not found
337 self
.logger
.debug(logging_text
+ "RO_vim_id={} already detached".format(RO_vim_id
))
341 step
= "Deleting vim from RO"
343 await RO
.delete("vim", RO_vim_id
)
344 except ROclient
.ROClientException
as e
:
345 if e
.http_code
== 404: # not found
346 self
.logger
.debug(logging_text
+ "RO_vim_id={} already deleted".format(RO_vim_id
))
351 self
.logger
.error(logging_text
+ "Nothing to remove at RO")
352 self
.db
.del_one("vim_accounts", {"_id": vim_id
})
354 self
.logger
.debug(logging_text
+ "Exit Ok")
357 except (ROclient
.ROClientException
, DbException
) as e
:
358 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
360 except Exception as e
:
361 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
364 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
366 db_vim_update
["_admin.operationalState"] = "ERROR"
367 db_vim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
368 # Mark the VIM 'delete' HA task as erroneous
369 operationState_HA
= 'FAILED'
370 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
371 self
.lcm_tasks
.register_HA('vim', 'delete', op_id
,
372 operationState
=operationState_HA
,
373 detailed_status
=detailed_status_HA
)
375 if db_vim
and db_vim_update
:
376 self
.update_db_2("vim_accounts", vim_id
, db_vim_update
)
377 # If the VIM 'delete' HA task was succesful, the DB entry has been deleted,
378 # which means that there is nowhere to register this task, so do nothing here.
379 except DbException
as e
:
380 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
381 self
.lcm_tasks
.remove("vim_account", vim_id
, order_id
)
384 class WimLcm(LcmBase
):
385 # values that are encrypted at wim config because they are passwords
386 wim_config_encrypted
= ()
388 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
390 Init, Connect to database, filesystem storage, and messaging
391 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
395 self
.logger
= logging
.getLogger('lcm.vim')
397 self
.lcm_tasks
= lcm_tasks
398 self
.ro_config
= config
["ro_config"]
400 super().__init
__(db
, msg
, fs
, self
.logger
)
402 async def create(self
, wim_content
, order_id
):
404 # HA tasks and backward compatibility:
405 # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
406 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
407 # Register 'create' task here for related future HA operations
408 op_id
= wim_content
.pop('op_id', None)
409 self
.lcm_tasks
.lock_HA('wim', 'create', op_id
)
411 wim_id
= wim_content
["_id"]
412 wim_content
.pop("op_id", None)
413 logging_text
= "Task wim_create={} ".format(wim_id
)
414 self
.logger
.debug(logging_text
+ "Enter")
419 operationState_HA
= ''
420 detailed_status_HA
= ''
422 step
= "Getting wim-id='{}' from db".format(wim_id
)
423 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
424 db_wim_update
["_admin.deployed.RO"] = None
426 step
= "Creating wim at RO"
427 db_wim_update
["_admin.detailed-status"] = step
428 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
429 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
430 wim_RO
= deepcopy(wim_content
)
431 wim_RO
.pop("_id", None)
432 wim_RO
.pop("_admin", None)
433 schema_version
= wim_RO
.pop("schema_version", None)
434 wim_RO
.pop("schema_type", None)
435 wim_RO
.pop("wim_tenant_name", None)
436 wim_RO
["type"] = wim_RO
.pop("wim_type")
437 wim_RO
.pop("wim_user", None)
438 wim_RO
.pop("wim_password", None)
439 desc
= await RO
.create("wim", descriptor
=wim_RO
)
440 RO_wim_id
= desc
["uuid"]
441 db_wim_update
["_admin.deployed.RO"] = RO_wim_id
442 self
.logger
.debug(logging_text
+ "WIM created at RO_wim_id={}".format(RO_wim_id
))
444 step
= "Creating wim_account at RO"
445 db_wim_update
["_admin.detailed-status"] = step
446 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
448 if wim_content
.get("wim_password"):
449 wim_content
["wim_password"] = self
.db
.decrypt(wim_content
["wim_password"],
450 schema_version
=schema_version
,
452 wim_account_RO
= {"name": wim_content
["name"],
453 "user": wim_content
["user"],
454 "password": wim_content
["password"]
456 if wim_RO
.get("config"):
457 wim_account_RO
["config"] = wim_RO
["config"]
458 if "wim_port_mapping" in wim_account_RO
["config"]:
459 del wim_account_RO
["config"]["wim_port_mapping"]
460 for p
in self
.wim_config_encrypted
:
461 if wim_account_RO
["config"].get(p
):
462 wim_account_RO
["config"][p
] = self
.db
.decrypt(wim_account_RO
["config"][p
],
463 schema_version
=schema_version
,
466 desc
= await RO
.attach("wim_account", RO_wim_id
, descriptor
=wim_account_RO
)
467 db_wim_update
["_admin.deployed.RO-account"] = desc
["uuid"]
468 db_wim_update
["_admin.operationalState"] = "ENABLED"
469 db_wim_update
["_admin.detailed-status"] = "Done"
470 # Mark the WIM 'create' HA task as successful
471 operationState_HA
= 'COMPLETED'
472 detailed_status_HA
= 'Done'
474 self
.logger
.debug(logging_text
+ "Exit Ok WIM account created at RO_wim_account_id={}".format(desc
["uuid"]))
477 except (ROclient
.ROClientException
, DbException
) as e
:
478 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
480 except Exception as e
:
481 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
485 db_wim_update
["_admin.operationalState"] = "ERROR"
486 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
487 # Mark the WIM 'create' HA task as erroneous
488 operationState_HA
= 'FAILED'
489 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
492 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
493 # Register the WIM 'create' HA task either
494 # succesful or erroneous, or do nothing (if legacy NBI)
495 self
.lcm_tasks
.register_HA('wim', 'create', op_id
,
496 operationState
=operationState_HA
,
497 detailed_status
=detailed_status_HA
)
498 except DbException
as e
:
499 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
500 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
502 async def edit(self
, wim_content
, order_id
):
504 # HA tasks and backward compatibility:
505 # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
506 # In such a case, HA is not supported by NBI, and the HA check always returns True
507 op_id
= wim_content
.pop('op_id', None)
508 if not self
.lcm_tasks
.lock_HA('wim', 'edit', op_id
):
511 wim_id
= wim_content
["_id"]
512 wim_content
.pop("op_id", None)
513 logging_text
= "Task wim_edit={} ".format(wim_id
)
514 self
.logger
.debug(logging_text
+ "Enter")
520 step
= "Getting wim-id='{}' from db".format(wim_id
)
521 operationState_HA
= ''
522 detailed_status_HA
= ''
524 # wait for any previous tasks in process
525 await self
.lcm_tasks
.waitfor_related_HA('wim', 'edit', op_id
)
527 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
529 if db_wim
.get("_admin") and db_wim
["_admin"].get("deployed") and db_wim
["_admin"]["deployed"].get("RO"):
531 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO"]
532 step
= "Editing wim at RO"
533 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
534 wim_RO
= deepcopy(wim_content
)
535 wim_RO
.pop("_id", None)
536 wim_RO
.pop("_admin", None)
537 schema_version
= wim_RO
.pop("schema_version", None)
538 wim_RO
.pop("schema_type", None)
539 wim_RO
.pop("wim_tenant_name", None)
540 if "wim_type" in wim_RO
:
541 wim_RO
["type"] = wim_RO
.pop("wim_type")
542 wim_RO
.pop("wim_user", None)
543 wim_RO
.pop("wim_password", None)
544 # TODO make a deep update of wim_port_mapping
546 await RO
.edit("wim", RO_wim_id
, descriptor
=wim_RO
)
548 step
= "Editing wim-account at RO tenant"
550 if "config" in wim_content
:
551 if "wim_port_mapping" in wim_content
["config"]:
552 del wim_content
["config"]["wim_port_mapping"]
553 if not wim_content
["config"]:
554 del wim_content
["config"]
555 if "wim_tenant_name" in wim_content
:
556 wim_account_RO
["wim_tenant_name"] = wim_content
["wim_tenant_name"]
557 if "wim_password" in wim_content
:
558 wim_account_RO
["wim_password"] = wim_content
["wim_password"]
559 if wim_content
.get("wim_password"):
560 wim_account_RO
["wim_password"] = self
.db
.decrypt(wim_content
["wim_password"],
561 schema_version
=schema_version
,
563 if "config" in wim_content
:
564 wim_account_RO
["config"] = wim_content
["config"]
565 if wim_content
.get("config"):
566 for p
in self
.wim_config_encrypted
:
567 if wim_content
["config"].get(p
):
568 wim_account_RO
["config"][p
] = self
.db
.decrypt(wim_content
["config"][p
],
569 schema_version
=schema_version
,
572 if "wim_user" in wim_content
:
573 wim_content
["wim_username"] = wim_content
["wim_user"]
574 # wim_account must be edited always even if empty in order to ensure changes are translated to RO
575 # wim_thread. RO will remove and relaunch a new thread for this wim_account
576 await RO
.edit("wim_account", RO_wim_id
, descriptor
=wim_account_RO
)
577 db_wim_update
["_admin.operationalState"] = "ENABLED"
578 # Mark the WIM 'edit' HA task as successful
579 operationState_HA
= 'COMPLETED'
580 detailed_status_HA
= 'Done'
582 self
.logger
.debug(logging_text
+ "Exit Ok RO_wim_id={}".format(RO_wim_id
))
585 except (ROclient
.ROClientException
, DbException
) as e
:
586 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
588 except Exception as e
:
589 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
593 db_wim_update
["_admin.operationalState"] = "ERROR"
594 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
595 # Mark the WIM 'edit' HA task as erroneous
596 operationState_HA
= 'FAILED'
597 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
600 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
601 # Register the WIM 'edit' HA task either
602 # succesful or erroneous, or do nothing (if legacy NBI)
603 self
.lcm_tasks
.register_HA('wim', 'edit', op_id
,
604 operationState
=operationState_HA
,
605 detailed_status
=detailed_status_HA
)
606 except DbException
as e
:
607 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
608 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
610 async def delete(self
, wim_content
, order_id
):
612 # HA tasks and backward compatibility:
613 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
614 # In such a case, HA is not supported by NBI, and the HA check always returns True
615 op_id
= wim_content
.pop('op_id', None)
616 if not self
.lcm_tasks
.lock_HA('wim', 'delete', op_id
):
619 wim_id
= wim_content
["_id"]
620 logging_text
= "Task wim_delete={} ".format(wim_id
)
621 self
.logger
.debug(logging_text
+ "Enter")
626 step
= "Getting wim from db"
627 operationState_HA
= ''
628 detailed_status_HA
= ''
630 # wait for any previous tasks in process
631 await self
.lcm_tasks
.waitfor_related_HA('wim', 'delete', op_id
)
633 db_wim
= self
.db
.get_one("wim_accounts", {"_id": wim_id
})
634 if db_wim
.get("_admin") and db_wim
["_admin"].get("deployed") and db_wim
["_admin"]["deployed"].get("RO"):
635 RO_wim_id
= db_wim
["_admin"]["deployed"]["RO"]
636 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
637 step
= "Detaching wim from RO tenant"
639 await RO
.detach("wim_account", RO_wim_id
)
640 except ROclient
.ROClientException
as e
:
641 if e
.http_code
== 404: # not found
642 self
.logger
.debug(logging_text
+ "RO_wim_id={} already detached".format(RO_wim_id
))
646 step
= "Deleting wim from RO"
648 await RO
.delete("wim", RO_wim_id
)
649 except ROclient
.ROClientException
as e
:
650 if e
.http_code
== 404: # not found
651 self
.logger
.debug(logging_text
+ "RO_wim_id={} already deleted".format(RO_wim_id
))
656 self
.logger
.error(logging_text
+ "Nohing to remove at RO")
657 self
.db
.del_one("wim_accounts", {"_id": wim_id
})
659 self
.logger
.debug(logging_text
+ "Exit Ok")
662 except (ROclient
.ROClientException
, DbException
) as e
:
663 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
665 except Exception as e
:
666 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
669 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
671 db_wim_update
["_admin.operationalState"] = "ERROR"
672 db_wim_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
673 # Mark the WIM 'delete' HA task as erroneous
674 operationState_HA
= 'FAILED'
675 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
676 self
.lcm_tasks
.register_HA('wim', 'delete', op_id
,
677 operationState
=operationState_HA
,
678 detailed_status
=detailed_status_HA
)
680 if db_wim
and db_wim_update
:
681 self
.update_db_2("wim_accounts", wim_id
, db_wim_update
)
682 # If the WIM 'delete' HA task was succesful, the DB entry has been deleted,
683 # which means that there is nowhere to register this task, so do nothing here.
684 except DbException
as e
:
685 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
686 self
.lcm_tasks
.remove("wim_account", wim_id
, order_id
)
689 class SdnLcm(LcmBase
):
691 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
693 Init, Connect to database, filesystem storage, and messaging
694 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
698 self
.logger
= logging
.getLogger('lcm.sdn')
700 self
.lcm_tasks
= lcm_tasks
701 self
.ro_config
= config
["ro_config"]
703 super().__init
__(db
, msg
, fs
, self
.logger
)
705 async def create(self
, sdn_content
, order_id
):
707 # HA tasks and backward compatibility:
708 # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
709 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
710 # Register 'create' task here for related future HA operations
711 op_id
= sdn_content
.pop('op_id', None)
712 self
.lcm_tasks
.lock_HA('sdn', 'create', op_id
)
714 sdn_id
= sdn_content
["_id"]
715 sdn_content
.pop("op_id", None)
716 logging_text
= "Task sdn_create={} ".format(sdn_id
)
717 self
.logger
.debug(logging_text
+ "Enter")
723 operationState_HA
= ''
724 detailed_status_HA
= ''
726 step
= "Getting sdn from db"
727 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
728 db_sdn_update
["_admin.deployed.RO"] = None
730 step
= "Creating sdn at RO"
731 db_sdn_update
["_admin.detailed-status"] = step
732 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
734 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
735 sdn_RO
= deepcopy(sdn_content
)
736 sdn_RO
.pop("_id", None)
737 sdn_RO
.pop("_admin", None)
738 schema_version
= sdn_RO
.pop("schema_version", None)
739 sdn_RO
.pop("schema_type", None)
740 sdn_RO
.pop("description", None)
741 if sdn_RO
.get("password"):
742 sdn_RO
["password"] = self
.db
.decrypt(sdn_RO
["password"], schema_version
=schema_version
, salt
=sdn_id
)
744 desc
= await RO
.create("sdn", descriptor
=sdn_RO
)
745 RO_sdn_id
= desc
["uuid"]
746 db_sdn_update
["_admin.deployed.RO"] = RO_sdn_id
747 db_sdn_update
["_admin.operationalState"] = "ENABLED"
748 self
.logger
.debug(logging_text
+ "Exit Ok RO_sdn_id={}".format(RO_sdn_id
))
749 # Mark the SDN 'create' HA task as successful
750 operationState_HA
= 'COMPLETED'
751 detailed_status_HA
= 'Done'
754 except (ROclient
.ROClientException
, DbException
) as e
:
755 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
757 except Exception as e
:
758 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
762 db_sdn_update
["_admin.operationalState"] = "ERROR"
763 db_sdn_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
764 # Mark the SDN 'create' HA task as erroneous
765 operationState_HA
= 'FAILED'
766 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
768 if db_sdn
and db_sdn_update
:
769 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
770 # Register the SDN 'create' HA task either
771 # succesful or erroneous, or do nothing (if legacy NBI)
772 self
.lcm_tasks
.register_HA('sdn', 'create', op_id
,
773 operationState
=operationState_HA
,
774 detailed_status
=detailed_status_HA
)
775 except DbException
as e
:
776 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
777 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
779 async def edit(self
, sdn_content
, order_id
):
781 # HA tasks and backward compatibility:
782 # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
783 # In such a case, HA is not supported by NBI, and the HA check always returns True
784 op_id
= sdn_content
.pop('op_id', None)
785 if not self
.lcm_tasks
.lock_HA('sdn', 'edit', op_id
):
788 sdn_id
= sdn_content
["_id"]
789 sdn_content
.pop("op_id", None)
790 logging_text
= "Task sdn_edit={} ".format(sdn_id
)
791 self
.logger
.debug(logging_text
+ "Enter")
796 operationState_HA
= ''
797 detailed_status_HA
= ''
798 step
= "Getting sdn from db"
800 # wait for any previous tasks in process
801 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'edit', op_id
)
803 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
805 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
806 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
807 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
808 step
= "Editing sdn at RO"
809 sdn_RO
= deepcopy(sdn_content
)
810 sdn_RO
.pop("_id", None)
811 sdn_RO
.pop("_admin", None)
812 schema_version
= sdn_RO
.pop("schema_version", None)
813 sdn_RO
.pop("schema_type", None)
814 sdn_RO
.pop("description", None)
815 if sdn_RO
.get("password"):
816 sdn_RO
["password"] = self
.db
.decrypt(sdn_RO
["password"], schema_version
=schema_version
, salt
=sdn_id
)
818 await RO
.edit("sdn", RO_sdn_id
, descriptor
=sdn_RO
)
819 db_sdn_update
["_admin.operationalState"] = "ENABLED"
820 # Mark the SDN 'edit' HA task as successful
821 operationState_HA
= 'COMPLETED'
822 detailed_status_HA
= 'Done'
824 self
.logger
.debug(logging_text
+ "Exit Ok RO_sdn_id={}".format(RO_sdn_id
))
827 except (ROclient
.ROClientException
, DbException
) as e
:
828 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
830 except Exception as e
:
831 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
835 db_sdn
["_admin.operationalState"] = "ERROR"
836 db_sdn
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
837 # Mark the SDN 'edit' HA task as erroneous
838 operationState_HA
= 'FAILED'
839 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
842 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
843 # Register the SDN 'edit' HA task either
844 # succesful or erroneous, or do nothing (if legacy NBI)
845 self
.lcm_tasks
.register_HA('sdn', 'edit', op_id
,
846 operationState
=operationState_HA
,
847 detailed_status
=detailed_status_HA
)
848 except DbException
as e
:
849 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
850 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
852 async def delete(self
, sdn_content
, order_id
):
854 # HA tasks and backward compatibility:
855 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
856 # In such a case, HA is not supported by NBI, and the HA check always returns True
857 op_id
= sdn_content
.pop('op_id', None)
858 if not self
.lcm_tasks
.lock_HA('sdn', 'delete', op_id
):
861 sdn_id
= sdn_content
["_id"]
862 logging_text
= "Task sdn_delete={} ".format(sdn_id
)
863 self
.logger
.debug(logging_text
+ "Enter")
868 operationState_HA
= ''
869 detailed_status_HA
= ''
870 step
= "Getting sdn from db"
872 # wait for any previous tasks in process
873 await self
.lcm_tasks
.waitfor_related_HA('sdn', 'delete', op_id
)
875 db_sdn
= self
.db
.get_one("sdns", {"_id": sdn_id
})
876 if db_sdn
.get("_admin") and db_sdn
["_admin"].get("deployed") and db_sdn
["_admin"]["deployed"].get("RO"):
877 RO_sdn_id
= db_sdn
["_admin"]["deployed"]["RO"]
878 RO
= ROclient
.ROClient(self
.loop
, **self
.ro_config
)
879 step
= "Deleting sdn from RO"
881 await RO
.delete("sdn", RO_sdn_id
)
882 except ROclient
.ROClientException
as e
:
883 if e
.http_code
== 404: # not found
884 self
.logger
.debug(logging_text
+ "RO_sdn_id={} already deleted".format(RO_sdn_id
))
889 self
.logger
.error(logging_text
+ "Skipping. There is not RO information at database")
890 self
.db
.del_one("sdns", {"_id": sdn_id
})
892 self
.logger
.debug("sdn_delete task sdn_id={} Exit Ok".format(sdn_id
))
895 except (ROclient
.ROClientException
, DbException
) as e
:
896 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
898 except Exception as e
:
899 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
903 db_sdn
["_admin.operationalState"] = "ERROR"
904 db_sdn
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
905 # Mark the SDN 'delete' HA task as erroneous
906 operationState_HA
= 'FAILED'
907 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
908 self
.lcm_tasks
.register_HA('sdn', 'delete', op_id
,
909 operationState
=operationState_HA
,
910 detailed_status
=detailed_status_HA
)
912 if db_sdn
and db_sdn_update
:
913 self
.update_db_2("sdns", sdn_id
, db_sdn_update
)
914 # If the SDN 'delete' HA task was succesful, the DB entry has been deleted,
915 # which means that there is nowhere to register this task, so do nothing here.
916 except DbException
as e
:
917 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
918 self
.lcm_tasks
.remove("sdn", sdn_id
, order_id
)
921 class K8sClusterLcm(LcmBase
):
923 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
925 Init, Connect to database, filesystem storage, and messaging
926 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
930 self
.logger
= logging
.getLogger('lcm.k8scluster')
932 self
.lcm_tasks
= lcm_tasks
933 self
.vca_config
= config
["VCA"]
937 self
.helm_k8scluster
= K8sHelmConnector(
938 kubectl_command
=self
.vca_config
.get("kubectlpath"),
939 helm_command
=self
.vca_config
.get("helmpath"),
946 self
.juju_k8scluster
= K8sJujuConnector(
947 kubectl_command
=self
.vca_config
.get("kubectlpath"),
948 juju_command
=self
.vca_config
.get("jujupath"),
955 super().__init
__(db
, msg
, fs
, self
.logger
)
957 async def create(self
, k8scluster_content
, order_id
):
959 # HA tasks and backward compatibility:
960 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
961 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
962 # Register 'create' task here for related future HA operations
963 op_id
= k8scluster_content
.pop('op_id', None)
964 if not self
.lcm_tasks
.lock_HA('k8scluster', 'create', op_id
):
967 k8scluster_id
= k8scluster_content
["_id"]
968 k8scluster_content
.pop("op_id", None)
969 logging_text
= "Task k8scluster_create={} ".format(k8scluster_id
)
970 self
.logger
.debug(logging_text
+ "Enter")
973 db_k8scluster_update
= {}
976 operationState_HA
= ''
977 detailed_status_HA
= ''
979 step
= "Getting k8scluster-id='{}' from db".format(k8scluster_id
)
980 self
.logger
.debug(logging_text
+ step
)
981 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": k8scluster_id
})
982 self
.db
.encrypt_decrypt_fields(db_k8scluster
.get("credentials"), 'decrypt', ['password', 'secret'],
983 schema_version
=db_k8scluster
["schema_version"], salt
=db_k8scluster
["_id"])
984 k8s_credentials
= yaml
.safe_dump(db_k8scluster
.get("credentials"))
989 k8s_hc_id
, uninstall_sw
= await self
.helm_k8scluster
.init_env(k8s_credentials
)
990 db_k8scluster_update
["_admin.helm-chart.id"] = k8s_hc_id
991 db_k8scluster_update
["_admin.helm-chart.created"] = uninstall_sw
992 except Exception as e
:
993 error_text_list
.append("Failing init helm-chart: {}".format(e
))
994 db_k8scluster_update
["_admin.helm-chart.error_msg"] = str(e
)
995 if isinstance(e
, K8sException
):
996 self
.logger
.error(logging_text
+ "Failing init helm-chart: {}".format(e
))
998 self
.logger
.error(logging_text
+ "Failing init helm-chart: {}".format(e
), exc_info
=True)
1003 k8s_jb_id
, uninstall_sw
= await self
.juju_k8scluster
.init_env(k8s_credentials
)
1004 db_k8scluster_update
["_admin.juju-bundle.id"] = k8s_jb_id
1005 db_k8scluster_update
["_admin.juju-bundle.created"] = uninstall_sw
1006 except Exception as e
:
1007 error_text_list
.append("Failing init juju-bundle: {}".format(e
))
1008 db_k8scluster_update
["_admin.juju-bundle.error_msg"] = str(e
)
1009 if isinstance(e
, N2VCException
):
1010 self
.logger
.error(logging_text
+ "Failing init juju-bundle: {}".format(e
))
1012 self
.logger
.error(logging_text
+ "Failing init juju-bundle: {}".format(e
), exc_info
=True)
1014 # mark as an error if both helm-chart and juju-bundle have been failed
1015 if k8s_hc_id
or k8s_jb_id
:
1016 self
.logger
.debug(logging_text
+ " successfully created")
1017 db_k8scluster_update
["_admin.operationalState"] = "ENABLED"
1019 self
.logger
.debug(logging_text
+ " successfully created with errors")
1020 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1021 db_k8scluster_update
["_admin.detailed-status"] = ";".join(error_text_list
)
1023 except Exception as e
:
1024 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1027 if exc
and db_k8scluster
:
1028 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1029 db_k8scluster_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1031 # Mark the k8scluster 'create' HA task as erroneous
1032 operationState_HA
= 'FAILED'
1033 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
1035 if db_k8scluster_update
:
1036 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1038 # Register the K8scluster 'create' HA task either
1039 # succesful or erroneous, or do nothing (if legacy NBI)
1040 self
.lcm_tasks
.register_HA('k8scluster', 'create', op_id
,
1041 operationState
=operationState_HA
,
1042 detailed_status
=detailed_status_HA
)
1043 except DbException
as e
:
1044 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1045 self
.lcm_tasks
.remove("k8scluster", k8scluster_id
, order_id
)
1047 async def delete(self
, k8scluster_content
, order_id
):
1049 # HA tasks and backward compatibility:
1050 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1051 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1052 # Register 'delete' task here for related future HA operations
1053 op_id
= k8scluster_content
.pop('op_id', None)
1054 if not self
.lcm_tasks
.lock_HA('k8scluster', 'delete', op_id
):
1057 k8scluster_id
= k8scluster_content
["_id"]
1058 k8scluster_content
.pop("op_id", None)
1059 logging_text
= "Task k8scluster_delete={} ".format(k8scluster_id
)
1060 self
.logger
.debug(logging_text
+ "Enter")
1062 db_k8scluster
= None
1063 db_k8scluster_update
= {}
1065 operationState_HA
= ''
1066 detailed_status_HA
= ''
1068 step
= "Getting k8scluster='{}' from db".format(k8scluster_id
)
1069 self
.logger
.debug(logging_text
+ step
)
1070 db_k8scluster
= self
.db
.get_one("k8sclusters", {"_id": k8scluster_id
})
1071 k8s_hc_id
= deep_get(db_k8scluster
, ("_admin", "helm-chart", "id"))
1072 k8s_jb_id
= deep_get(db_k8scluster
, ("_admin", "juju-bundle", "id"))
1074 uninstall_sw
= deep_get(db_k8scluster
, ("_admin", "helm-chart", "created"))
1075 cluster_removed
= True
1077 uninstall_sw
= uninstall_sw
or False
1078 cluster_removed
= await self
.helm_k8scluster
.reset(cluster_uuid
=k8s_hc_id
, uninstall_sw
=uninstall_sw
)
1081 uninstall_sw
= uninstall_sw
or False
1082 cluster_removed
= await self
.juju_k8scluster
.reset(cluster_uuid
=k8s_jb_id
, uninstall_sw
=uninstall_sw
)
1084 # Try to remove from cluster_inserted to clean old versions
1085 if k8s_hc_id
and cluster_removed
:
1086 step
= "Removing k8scluster='{}' from k8srepos".format(k8scluster_id
)
1087 self
.logger
.debug(logging_text
+ step
)
1088 db_k8srepo_list
= self
.db
.get_list("k8srepos", {"_admin.cluster-inserted": k8s_hc_id
})
1089 for k8srepo
in db_k8srepo_list
:
1091 cluster_list
= k8srepo
["_admin"]["cluster-inserted"]
1092 cluster_list
.remove(k8s_hc_id
)
1093 self
.update_db_2("k8srepos", k8srepo
["_id"], {"_admin.cluster-inserted": cluster_list
})
1094 except Exception as e
:
1095 self
.logger
.error("{}: {}".format(step
, e
))
1096 self
.db
.del_one("k8sclusters", {"_id": k8scluster_id
})
1098 raise LcmException("An error happened during the reset of the k8s cluster '{}'".format(k8scluster_id
))
1099 # if not cluster_removed:
1100 # raise Exception("K8scluster was not properly removed")
1102 except Exception as e
:
1103 if isinstance(e
, (LcmException
, DbException
)):
1104 self
.logger
.error(logging_text
+ "Exit Exception {}".format(e
))
1106 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1109 if exc
and db_k8scluster
:
1110 db_k8scluster_update
["_admin.operationalState"] = "ERROR"
1111 db_k8scluster_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1112 # Mark the WIM 'create' HA task as erroneous
1113 operationState_HA
= 'FAILED'
1114 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
1116 if db_k8scluster_update
:
1117 self
.update_db_2("k8sclusters", k8scluster_id
, db_k8scluster_update
)
1118 # Register the K8scluster 'delete' HA task either
1119 # succesful or erroneous, or do nothing (if legacy NBI)
1120 self
.lcm_tasks
.register_HA('k8scluster', 'delete', op_id
,
1121 operationState
=operationState_HA
,
1122 detailed_status
=detailed_status_HA
)
1123 except DbException
as e
:
1124 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1125 self
.lcm_tasks
.remove("k8scluster", k8scluster_id
, order_id
)
1128 class K8sRepoLcm(LcmBase
):
1130 def __init__(self
, db
, msg
, fs
, lcm_tasks
, config
, loop
):
1132 Init, Connect to database, filesystem storage, and messaging
1133 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1137 self
.logger
= logging
.getLogger('lcm.k8srepo')
1139 self
.lcm_tasks
= lcm_tasks
1140 self
.vca_config
= config
["VCA"]
1144 self
.k8srepo
= K8sHelmConnector(
1145 kubectl_command
=self
.vca_config
.get("kubectlpath"),
1146 helm_command
=self
.vca_config
.get("helmpath"),
1153 super().__init
__(db
, msg
, fs
, self
.logger
)
1155 async def create(self
, k8srepo_content
, order_id
):
1157 # HA tasks and backward compatibility:
1158 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1159 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1160 # Register 'create' task here for related future HA operations
1162 op_id
= k8srepo_content
.pop('op_id', None)
1163 if not self
.lcm_tasks
.lock_HA('k8srepo', 'create', op_id
):
1166 k8srepo_id
= k8srepo_content
.get("_id")
1167 logging_text
= "Task k8srepo_create={} ".format(k8srepo_id
)
1168 self
.logger
.debug(logging_text
+ "Enter")
1171 db_k8srepo_update
= {}
1173 operationState_HA
= ''
1174 detailed_status_HA
= ''
1176 step
= "Getting k8srepo-id='{}' from db".format(k8srepo_id
)
1177 self
.logger
.debug(logging_text
+ step
)
1178 db_k8srepo
= self
.db
.get_one("k8srepos", {"_id": k8srepo_id
})
1179 db_k8srepo_update
["_admin.operationalState"] = "ENABLED"
1180 except Exception as e
:
1181 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1184 if exc
and db_k8srepo
:
1185 db_k8srepo_update
["_admin.operationalState"] = "ERROR"
1186 db_k8srepo_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1187 # Mark the WIM 'create' HA task as erroneous
1188 operationState_HA
= 'FAILED'
1189 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
1191 if db_k8srepo_update
:
1192 self
.update_db_2("k8srepos", k8srepo_id
, db_k8srepo_update
)
1193 # Register the K8srepo 'create' HA task either
1194 # succesful or erroneous, or do nothing (if legacy NBI)
1195 self
.lcm_tasks
.register_HA('k8srepo', 'create', op_id
,
1196 operationState
=operationState_HA
,
1197 detailed_status
=detailed_status_HA
)
1198 except DbException
as e
:
1199 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1200 self
.lcm_tasks
.remove("k8srepo", k8srepo_id
, order_id
)
1202 async def delete(self
, k8srepo_content
, order_id
):
1204 # HA tasks and backward compatibility:
1205 # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1206 # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1207 # Register 'delete' task here for related future HA operations
1208 op_id
= k8srepo_content
.pop('op_id', None)
1209 if not self
.lcm_tasks
.lock_HA('k8srepo', 'delete', op_id
):
1212 k8srepo_id
= k8srepo_content
.get("_id")
1213 logging_text
= "Task k8srepo_delete={} ".format(k8srepo_id
)
1214 self
.logger
.debug(logging_text
+ "Enter")
1217 db_k8srepo_update
= {}
1220 operationState_HA
= ''
1221 detailed_status_HA
= ''
1223 step
= "Getting k8srepo-id='{}' from db".format(k8srepo_id
)
1224 self
.logger
.debug(logging_text
+ step
)
1225 db_k8srepo
= self
.db
.get_one("k8srepos", {"_id": k8srepo_id
})
1227 except Exception as e
:
1228 self
.logger
.critical(logging_text
+ "Exit Exception {}".format(e
), exc_info
=True)
1231 if exc
and db_k8srepo
:
1232 db_k8srepo_update
["_admin.operationalState"] = "ERROR"
1233 db_k8srepo_update
["_admin.detailed-status"] = "ERROR {}: {}".format(step
, exc
)
1234 # Mark the WIM 'create' HA task as erroneous
1235 operationState_HA
= 'FAILED'
1236 detailed_status_HA
= "ERROR {}: {}".format(step
, exc
)
1238 if db_k8srepo_update
:
1239 self
.update_db_2("k8srepos", k8srepo_id
, db_k8srepo_update
)
1240 # Register the K8srepo 'delete' HA task either
1241 # succesful or erroneous, or do nothing (if legacy NBI)
1242 self
.lcm_tasks
.register_HA('k8srepo', 'delete', op_id
,
1243 operationState
=operationState_HA
,
1244 detailed_status
=detailed_status_HA
)
1245 self
.db
.del_one("k8srepos", {"_id": k8srepo_id
})
1246 except DbException
as e
:
1247 self
.logger
.error(logging_text
+ "Cannot update database: {}".format(e
))
1248 self
.lcm_tasks
.remove("k8srepo", k8srepo_id
, order_id
)