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