osm_lcm/vim_sdn.py

   1 # -*- coding: utf-8 -*-
   2
   3 ##
   4 # Copyright 2018 Telefonica S.A.
   5 #
   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
   9 #
  10 #         http://www.apache.org/licenses/LICENSE-2.0
  11 #
  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
  16 # under the License.
  17 ##
  18
  19 import yaml
  20 import asyncio
  21 import logging
  22 import logging.handlers
  23 from osm_lcm import ROclient
  24 from osm_lcm.lcm_utils import LcmException, LcmBase, deep_get
  25 from n2vc.k8s_helm_conn import K8sHelmConnector
  26 from n2vc.k8s_helm3_conn import K8sHelm3Connector
  27 from n2vc.k8s_juju_conn import K8sJujuConnector
  28 from n2vc.n2vc_juju_conn import N2VCJujuConnector
  29 from n2vc.exceptions import K8sException, N2VCException
  30 from osm_common.dbbase import DbException
  31 from copy import deepcopy
  32 from time import time
  33
  34 __author__ = "Alfonso Tierno"
  35
  36
  37 class VimLcm(LcmBase):
  38     # values that are encrypted at vim config because they are passwords
  39     vim_config_encrypted = {
  40         "1.1": ("admin_password", "nsx_password", "vcenter_password"),
  41         "default": (
  42             "admin_password",
  43             "nsx_password",
  44             "vcenter_password",
  45             "vrops_password",
  46         ),
  47     }
  48
  49     def __init__(self, msg, lcm_tasks, config, loop):
  50         """
  51         Init, Connect to database, filesystem storage, and messaging
  52         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
  53         :return: None
  54         """
  55
  56         self.logger = logging.getLogger("lcm.vim")
  57         self.loop = loop
  58         self.lcm_tasks = lcm_tasks
  59         self.ro_config = config["ro_config"]
  60
  61         super().__init__(msg, self.logger)
  62
  63     async def create(self, vim_content, order_id):
  64
  65         # HA tasks and backward compatibility:
  66         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
  67         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
  68         # Register 'create' task here for related future HA operations
  69         op_id = vim_content.pop("op_id", None)
  70         if not self.lcm_tasks.lock_HA("vim", "create", op_id):
  71             return
  72
  73         vim_id = vim_content["_id"]
  74         logging_text = "Task vim_create={} ".format(vim_id)
  75         self.logger.debug(logging_text + "Enter")
  76
  77         db_vim = None
  78         db_vim_update = {}
  79         exc = None
  80         RO_sdn_id = None
  81         try:
  82             step = "Getting vim-id='{}' from db".format(vim_id)
  83             db_vim = self.db.get_one("vim_accounts", {"_id": vim_id})
  84             if vim_content.get("config") and vim_content["config"].get(
  85                 "sdn-controller"
  86             ):
  87                 step = "Getting sdn-controller-id='{}' from db".format(
  88                     vim_content["config"]["sdn-controller"]
  89                 )
  90                 db_sdn = self.db.get_one(
  91                     "sdns", {"_id": vim_content["config"]["sdn-controller"]}
  92                 )
  93
  94                 # If the VIM account has an associated SDN account, also
  95                 # wait for any previous tasks in process for the SDN
  96                 await self.lcm_tasks.waitfor_related_HA("sdn", "ANY", db_sdn["_id"])
  97
  98                 if (
  99                     db_sdn.get("_admin")
 100                     and db_sdn["_admin"].get("deployed")
 101                     and db_sdn["_admin"]["deployed"].get("RO")
 102                 ):
 103                     RO_sdn_id = db_sdn["_admin"]["deployed"]["RO"]
 104                 else:
 105                     raise LcmException(
 106                         "sdn-controller={} is not available. Not deployed at RO".format(
 107                             vim_content["config"]["sdn-controller"]
 108                         )
 109                     )
 110
 111             step = "Creating vim at RO"
 112             db_vim_update["_admin.deployed.RO"] = None
 113             db_vim_update["_admin.detailed-status"] = step
 114             self.update_db_2("vim_accounts", vim_id, db_vim_update)
 115             RO = ROclient.ROClient(self.loop, **self.ro_config)
 116             vim_RO = deepcopy(vim_content)
 117             vim_RO.pop("_id", None)
 118             vim_RO.pop("_admin", None)
 119             schema_version = vim_RO.pop("schema_version", None)
 120             vim_RO.pop("schema_type", None)
 121             vim_RO.pop("vim_tenant_name", None)
 122             vim_RO["type"] = vim_RO.pop("vim_type")
 123             vim_RO.pop("vim_user", None)
 124             vim_RO.pop("vim_password", None)
 125             if RO_sdn_id:
 126                 vim_RO["config"]["sdn-controller"] = RO_sdn_id
 127             desc = await RO.create("vim", descriptor=vim_RO)
 128             RO_vim_id = desc["uuid"]
 129             db_vim_update["_admin.deployed.RO"] = RO_vim_id
 130             self.logger.debug(
 131                 logging_text + "VIM created at RO_vim_id={}".format(RO_vim_id)
 132             )
 133
 134             step = "Creating vim_account at RO"
 135             db_vim_update["_admin.detailed-status"] = step
 136             self.update_db_2("vim_accounts", vim_id, db_vim_update)
 137
 138             if vim_content.get("vim_password"):
 139                 vim_content["vim_password"] = self.db.decrypt(
 140                     vim_content["vim_password"],
 141                     schema_version=schema_version,
 142                     salt=vim_id,
 143                 )
 144             vim_account_RO = {
 145                 "vim_tenant_name": vim_content["vim_tenant_name"],
 146                 "vim_username": vim_content["vim_user"],
 147                 "vim_password": vim_content["vim_password"],
 148             }
 149             if vim_RO.get("config"):
 150                 vim_account_RO["config"] = vim_RO["config"]
 151                 if "sdn-controller" in vim_account_RO["config"]:
 152                     del vim_account_RO["config"]["sdn-controller"]
 153                 if "sdn-port-mapping" in vim_account_RO["config"]:
 154                     del vim_account_RO["config"]["sdn-port-mapping"]
 155                 vim_config_encrypted_keys = self.vim_config_encrypted.get(
 156                     schema_version
 157                 ) or self.vim_config_encrypted.get("default")
 158                 for p in vim_config_encrypted_keys:
 159                     if vim_account_RO["config"].get(p):
 160                         vim_account_RO["config"][p] = self.db.decrypt(
 161                             vim_account_RO["config"][p],
 162                             schema_version=schema_version,
 163                             salt=vim_id,
 164                         )
 165
 166             desc = await RO.attach("vim_account", RO_vim_id, descriptor=vim_account_RO)
 167             db_vim_update["_admin.deployed.RO-account"] = desc["uuid"]
 168             db_vim_update["_admin.operationalState"] = "ENABLED"
 169             db_vim_update["_admin.detailed-status"] = "Done"
 170             # Mark the VIM 'create' HA task as successful
 171             operation_state = "COMPLETED"
 172             operation_details = "Done"
 173
 174             self.logger.debug(
 175                 logging_text
 176                 + "Exit Ok VIM account created at RO_vim_account_id={}".format(
 177                     desc["uuid"]
 178                 )
 179             )
 180             return
 181
 182         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 183             self.logger.error(logging_text + "Exit Exception {}".format(e))
 184             exc = e
 185         except Exception as e:
 186             self.logger.critical(
 187                 logging_text + "Exit Exception {}".format(e), exc_info=True
 188             )
 189             exc = e
 190         finally:
 191             if exc and db_vim:
 192                 db_vim_update["_admin.operationalState"] = "ERROR"
 193                 db_vim_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 194                     step, exc
 195                 )
 196                 # Mark the VIM 'create' HA task as erroneous
 197                 operation_state = "FAILED"
 198                 operation_details = "ERROR {}: {}".format(step, exc)
 199             try:
 200                 if db_vim_update:
 201                     self.update_db_2("vim_accounts", vim_id, db_vim_update)
 202                 # Register the VIM 'create' HA task either
 203                 # succesful or erroneous, or do nothing (if legacy NBI)
 204                 self.lcm_tasks.unlock_HA(
 205                     "vim",
 206                     "create",
 207                     op_id,
 208                     operationState=operation_state,
 209                     detailed_status=operation_details,
 210                 )
 211             except DbException as e:
 212                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 213
 214             self.lcm_tasks.remove("vim_account", vim_id, order_id)
 215
 216     async def edit(self, vim_content, order_id):
 217
 218         # HA tasks and backward compatibility:
 219         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
 220         # In such a case, HA is not supported by NBI, and the HA check always returns True
 221         op_id = vim_content.pop("op_id", None)
 222         if not self.lcm_tasks.lock_HA("vim", "edit", op_id):
 223             return
 224
 225         vim_id = vim_content["_id"]
 226         logging_text = "Task vim_edit={} ".format(vim_id)
 227         self.logger.debug(logging_text + "Enter")
 228
 229         db_vim = None
 230         exc = None
 231         RO_sdn_id = None
 232         RO_vim_id = None
 233         db_vim_update = {}
 234         step = "Getting vim-id='{}' from db".format(vim_id)
 235         try:
 236             # wait for any previous tasks in process
 237             await self.lcm_tasks.waitfor_related_HA("vim", "edit", op_id)
 238
 239             db_vim = self.db.get_one("vim_accounts", {"_id": vim_id})
 240
 241             if (
 242                 db_vim.get("_admin")
 243                 and db_vim["_admin"].get("deployed")
 244                 and db_vim["_admin"]["deployed"].get("RO")
 245             ):
 246                 if vim_content.get("config") and vim_content["config"].get(
 247                     "sdn-controller"
 248                 ):
 249                     step = "Getting sdn-controller-id='{}' from db".format(
 250                         vim_content["config"]["sdn-controller"]
 251                     )
 252                     db_sdn = self.db.get_one(
 253                         "sdns", {"_id": vim_content["config"]["sdn-controller"]}
 254                     )
 255
 256                     # If the VIM account has an associated SDN account, also
 257                     # wait for any previous tasks in process for the SDN
 258                     await self.lcm_tasks.waitfor_related_HA("sdn", "ANY", db_sdn["_id"])
 259
 260                     if (
 261                         db_sdn.get("_admin")
 262                         and db_sdn["_admin"].get("deployed")
 263                         and db_sdn["_admin"]["deployed"].get("RO")
 264                     ):
 265                         RO_sdn_id = db_sdn["_admin"]["deployed"]["RO"]
 266                     else:
 267                         raise LcmException(
 268                             "sdn-controller={} is not available. Not deployed at RO".format(
 269                                 vim_content["config"]["sdn-controller"]
 270                             )
 271                         )
 272
 273                 RO_vim_id = db_vim["_admin"]["deployed"]["RO"]
 274                 step = "Editing vim at RO"
 275                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 276                 vim_RO = deepcopy(vim_content)
 277                 vim_RO.pop("_id", None)
 278                 vim_RO.pop("_admin", None)
 279                 schema_version = vim_RO.pop("schema_version", None)
 280                 vim_RO.pop("schema_type", None)
 281                 vim_RO.pop("vim_tenant_name", None)
 282                 if "vim_type" in vim_RO:
 283                     vim_RO["type"] = vim_RO.pop("vim_type")
 284                 vim_RO.pop("vim_user", None)
 285                 vim_RO.pop("vim_password", None)
 286                 if RO_sdn_id:
 287                     vim_RO["config"]["sdn-controller"] = RO_sdn_id
 288                 # TODO make a deep update of sdn-port-mapping
 289                 if vim_RO:
 290                     await RO.edit("vim", RO_vim_id, descriptor=vim_RO)
 291
 292                 step = "Editing vim-account at RO tenant"
 293                 vim_account_RO = {}
 294                 if "config" in vim_content:
 295                     if "sdn-controller" in vim_content["config"]:
 296                         del vim_content["config"]["sdn-controller"]
 297                     if "sdn-port-mapping" in vim_content["config"]:
 298                         del vim_content["config"]["sdn-port-mapping"]
 299                     if not vim_content["config"]:
 300                         del vim_content["config"]
 301                 if "vim_tenant_name" in vim_content:
 302                     vim_account_RO["vim_tenant_name"] = vim_content["vim_tenant_name"]
 303                 if "vim_password" in vim_content:
 304                     vim_account_RO["vim_password"] = vim_content["vim_password"]
 305                 if vim_content.get("vim_password"):
 306                     vim_account_RO["vim_password"] = self.db.decrypt(
 307                         vim_content["vim_password"],
 308                         schema_version=schema_version,
 309                         salt=vim_id,
 310                     )
 311                 if "config" in vim_content:
 312                     vim_account_RO["config"] = vim_content["config"]
 313                 if vim_content.get("config"):
 314                     vim_config_encrypted_keys = self.vim_config_encrypted.get(
 315                         schema_version
 316                     ) or self.vim_config_encrypted.get("default")
 317                     for p in vim_config_encrypted_keys:
 318                         if vim_content["config"].get(p):
 319                             vim_account_RO["config"][p] = self.db.decrypt(
 320                                 vim_content["config"][p],
 321                                 schema_version=schema_version,
 322                                 salt=vim_id,
 323                             )
 324
 325                 if "vim_user" in vim_content:
 326                     vim_content["vim_username"] = vim_content["vim_user"]
 327                 # vim_account must be edited always even if empty in order to ensure changes are translated to RO
 328                 # vim_thread. RO will remove and relaunch a new thread for this vim_account
 329                 await RO.edit("vim_account", RO_vim_id, descriptor=vim_account_RO)
 330                 db_vim_update["_admin.operationalState"] = "ENABLED"
 331                 # Mark the VIM 'edit' HA task as successful
 332             operation_state = "COMPLETED"
 333             operation_details = "Done"
 334
 335             self.logger.debug(logging_text + "Exit Ok RO_vim_id={}".format(RO_vim_id))
 336             return
 337
 338         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 339             self.logger.error(logging_text + "Exit Exception {}".format(e))
 340             exc = e
 341         except Exception as e:
 342             self.logger.critical(
 343                 logging_text + "Exit Exception {}".format(e), exc_info=True
 344             )
 345             exc = e
 346         finally:
 347             if exc and db_vim:
 348                 db_vim_update["_admin.operationalState"] = "ERROR"
 349                 db_vim_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 350                     step, exc
 351                 )
 352                 # Mark the VIM 'edit' HA task as erroneous
 353                 operation_state = "FAILED"
 354                 operation_details = "ERROR {}: {}".format(step, exc)
 355             try:
 356                 if db_vim_update:
 357                     self.update_db_2("vim_accounts", vim_id, db_vim_update)
 358                 # Register the VIM 'edit' HA task either
 359                 # succesful or erroneous, or do nothing (if legacy NBI)
 360                 self.lcm_tasks.unlock_HA(
 361                     "vim",
 362                     "edit",
 363                     op_id,
 364                     operationState=operation_state,
 365                     detailed_status=operation_details,
 366                 )
 367             except DbException as e:
 368                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 369
 370             self.lcm_tasks.remove("vim_account", vim_id, order_id)
 371
 372     async def delete(self, vim_content, order_id):
 373
 374         # HA tasks and backward compatibility:
 375         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
 376         # In such a case, HA is not supported by NBI, and the HA check always returns True
 377         op_id = vim_content.pop("op_id", None)
 378         if not self.lcm_tasks.lock_HA("vim", "delete", op_id):
 379             return
 380
 381         vim_id = vim_content["_id"]
 382         logging_text = "Task vim_delete={} ".format(vim_id)
 383         self.logger.debug(logging_text + "Enter")
 384
 385         db_vim = None
 386         db_vim_update = {}
 387         exc = None
 388         step = "Getting vim from db"
 389         try:
 390             # wait for any previous tasks in process
 391             await self.lcm_tasks.waitfor_related_HA("vim", "delete", op_id)
 392             if not self.ro_config.get("ng"):
 393                 db_vim = self.db.get_one("vim_accounts", {"_id": vim_id})
 394                 if (
 395                     db_vim.get("_admin")
 396                     and db_vim["_admin"].get("deployed")
 397                     and db_vim["_admin"]["deployed"].get("RO")
 398                 ):
 399                     RO_vim_id = db_vim["_admin"]["deployed"]["RO"]
 400                     RO = ROclient.ROClient(self.loop, **self.ro_config)
 401                     step = "Detaching vim from RO tenant"
 402                     try:
 403                         await RO.detach("vim_account", RO_vim_id)
 404                     except ROclient.ROClientException as e:
 405                         if e.http_code == 404:  # not found
 406                             self.logger.debug(
 407                                 logging_text
 408                                 + "RO_vim_id={} already detached".format(RO_vim_id)
 409                             )
 410                         else:
 411                             raise
 412
 413                     step = "Deleting vim from RO"
 414                     try:
 415                         await RO.delete("vim", RO_vim_id)
 416                     except ROclient.ROClientException as e:
 417                         if e.http_code == 404:  # not found
 418                             self.logger.debug(
 419                                 logging_text
 420                                 + "RO_vim_id={} already deleted".format(RO_vim_id)
 421                             )
 422                         else:
 423                             raise
 424                 else:
 425                     # nothing to delete
 426                     self.logger.debug(logging_text + "Nothing to remove at RO")
 427             self.db.del_one("vim_accounts", {"_id": vim_id})
 428             db_vim = None
 429             self.logger.debug(logging_text + "Exit Ok")
 430             return
 431
 432         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 433             self.logger.error(logging_text + "Exit Exception {}".format(e))
 434             exc = e
 435         except Exception as e:
 436             self.logger.critical(
 437                 logging_text + "Exit Exception {}".format(e), exc_info=True
 438             )
 439             exc = e
 440         finally:
 441             self.lcm_tasks.remove("vim_account", vim_id, order_id)
 442             if exc and db_vim:
 443                 db_vim_update["_admin.operationalState"] = "ERROR"
 444                 db_vim_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 445                     step, exc
 446                 )
 447                 # Mark the VIM 'delete' HA task as erroneous
 448                 operation_state = "FAILED"
 449                 operation_details = "ERROR {}: {}".format(step, exc)
 450                 self.lcm_tasks.unlock_HA(
 451                     "vim",
 452                     "delete",
 453                     op_id,
 454                     operationState=operation_state,
 455                     detailed_status=operation_details,
 456                 )
 457             try:
 458                 if db_vim and db_vim_update:
 459                     self.update_db_2("vim_accounts", vim_id, db_vim_update)
 460                 # If the VIM 'delete' HA task was succesful, the DB entry has been deleted,
 461                 # which means that there is nowhere to register this task, so do nothing here.
 462             except DbException as e:
 463                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 464             self.lcm_tasks.remove("vim_account", vim_id, order_id)
 465
 466
 467 class WimLcm(LcmBase):
 468     # values that are encrypted at wim config because they are passwords
 469     wim_config_encrypted = ()
 470
 471     def __init__(self, msg, lcm_tasks, config, loop):
 472         """
 473         Init, Connect to database, filesystem storage, and messaging
 474         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
 475         :return: None
 476         """
 477
 478         self.logger = logging.getLogger("lcm.vim")
 479         self.loop = loop
 480         self.lcm_tasks = lcm_tasks
 481         self.ro_config = config["ro_config"]
 482
 483         super().__init__(msg, self.logger)
 484
 485     async def create(self, wim_content, order_id):
 486
 487         # HA tasks and backward compatibility:
 488         # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
 489         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
 490         # Register 'create' task here for related future HA operations
 491         op_id = wim_content.pop("op_id", None)
 492         self.lcm_tasks.lock_HA("wim", "create", op_id)
 493
 494         wim_id = wim_content["_id"]
 495         logging_text = "Task wim_create={} ".format(wim_id)
 496         self.logger.debug(logging_text + "Enter")
 497
 498         db_wim = None
 499         db_wim_update = {}
 500         exc = None
 501         try:
 502             step = "Getting wim-id='{}' from db".format(wim_id)
 503             db_wim = self.db.get_one("wim_accounts", {"_id": wim_id})
 504             db_wim_update["_admin.deployed.RO"] = None
 505
 506             step = "Creating wim at RO"
 507             db_wim_update["_admin.detailed-status"] = step
 508             self.update_db_2("wim_accounts", wim_id, db_wim_update)
 509             RO = ROclient.ROClient(self.loop, **self.ro_config)
 510             wim_RO = deepcopy(wim_content)
 511             wim_RO.pop("_id", None)
 512             wim_RO.pop("_admin", None)
 513             schema_version = wim_RO.pop("schema_version", None)
 514             wim_RO.pop("schema_type", None)
 515             wim_RO.pop("wim_tenant_name", None)
 516             wim_RO["type"] = wim_RO.pop("wim_type")
 517             wim_RO.pop("wim_user", None)
 518             wim_RO.pop("wim_password", None)
 519             desc = await RO.create("wim", descriptor=wim_RO)
 520             RO_wim_id = desc["uuid"]
 521             db_wim_update["_admin.deployed.RO"] = RO_wim_id
 522             self.logger.debug(
 523                 logging_text + "WIM created at RO_wim_id={}".format(RO_wim_id)
 524             )
 525
 526             step = "Creating wim_account at RO"
 527             db_wim_update["_admin.detailed-status"] = step
 528             self.update_db_2("wim_accounts", wim_id, db_wim_update)
 529
 530             if wim_content.get("wim_password"):
 531                 wim_content["wim_password"] = self.db.decrypt(
 532                     wim_content["wim_password"],
 533                     schema_version=schema_version,
 534                     salt=wim_id,
 535                 )
 536             wim_account_RO = {
 537                 "name": wim_content["name"],
 538                 "user": wim_content["user"],
 539                 "password": wim_content["password"],
 540             }
 541             if wim_RO.get("config"):
 542                 wim_account_RO["config"] = wim_RO["config"]
 543                 if "wim_port_mapping" in wim_account_RO["config"]:
 544                     del wim_account_RO["config"]["wim_port_mapping"]
 545                 for p in self.wim_config_encrypted:
 546                     if wim_account_RO["config"].get(p):
 547                         wim_account_RO["config"][p] = self.db.decrypt(
 548                             wim_account_RO["config"][p],
 549                             schema_version=schema_version,
 550                             salt=wim_id,
 551                         )
 552
 553             desc = await RO.attach("wim_account", RO_wim_id, descriptor=wim_account_RO)
 554             db_wim_update["_admin.deployed.RO-account"] = desc["uuid"]
 555             db_wim_update["_admin.operationalState"] = "ENABLED"
 556             db_wim_update["_admin.detailed-status"] = "Done"
 557             # Mark the WIM 'create' HA task as successful
 558             operation_state = "COMPLETED"
 559             operation_details = "Done"
 560
 561             self.logger.debug(
 562                 logging_text
 563                 + "Exit Ok WIM account created at RO_wim_account_id={}".format(
 564                     desc["uuid"]
 565                 )
 566             )
 567             return
 568
 569         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 570             self.logger.error(logging_text + "Exit Exception {}".format(e))
 571             exc = e
 572         except Exception as e:
 573             self.logger.critical(
 574                 logging_text + "Exit Exception {}".format(e), exc_info=True
 575             )
 576             exc = e
 577         finally:
 578             if exc and db_wim:
 579                 db_wim_update["_admin.operationalState"] = "ERROR"
 580                 db_wim_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 581                     step, exc
 582                 )
 583                 # Mark the WIM 'create' HA task as erroneous
 584                 operation_state = "FAILED"
 585                 operation_details = "ERROR {}: {}".format(step, exc)
 586             try:
 587                 if db_wim_update:
 588                     self.update_db_2("wim_accounts", wim_id, db_wim_update)
 589                 # Register the WIM 'create' HA task either
 590                 # succesful or erroneous, or do nothing (if legacy NBI)
 591                 self.lcm_tasks.unlock_HA(
 592                     "wim",
 593                     "create",
 594                     op_id,
 595                     operationState=operation_state,
 596                     detailed_status=operation_details,
 597                 )
 598             except DbException as e:
 599                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 600             self.lcm_tasks.remove("wim_account", wim_id, order_id)
 601
 602     async def edit(self, wim_content, order_id):
 603
 604         # HA tasks and backward compatibility:
 605         # If 'wim_content' does not include 'op_id', we a running a legacy NBI version.
 606         # In such a case, HA is not supported by NBI, and the HA check always returns True
 607         op_id = wim_content.pop("op_id", None)
 608         if not self.lcm_tasks.lock_HA("wim", "edit", op_id):
 609             return
 610
 611         wim_id = wim_content["_id"]
 612         logging_text = "Task wim_edit={} ".format(wim_id)
 613         self.logger.debug(logging_text + "Enter")
 614
 615         db_wim = None
 616         exc = None
 617         RO_wim_id = None
 618         db_wim_update = {}
 619         step = "Getting wim-id='{}' from db".format(wim_id)
 620         try:
 621             # wait for any previous tasks in process
 622             await self.lcm_tasks.waitfor_related_HA("wim", "edit", op_id)
 623
 624             db_wim = self.db.get_one("wim_accounts", {"_id": wim_id})
 625
 626             if (
 627                 db_wim.get("_admin")
 628                 and db_wim["_admin"].get("deployed")
 629                 and db_wim["_admin"]["deployed"].get("RO")
 630             ):
 631
 632                 RO_wim_id = db_wim["_admin"]["deployed"]["RO"]
 633                 step = "Editing wim at RO"
 634                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 635                 wim_RO = deepcopy(wim_content)
 636                 wim_RO.pop("_id", None)
 637                 wim_RO.pop("_admin", None)
 638                 schema_version = wim_RO.pop("schema_version", None)
 639                 wim_RO.pop("schema_type", None)
 640                 wim_RO.pop("wim_tenant_name", None)
 641                 if "wim_type" in wim_RO:
 642                     wim_RO["type"] = wim_RO.pop("wim_type")
 643                 wim_RO.pop("wim_user", None)
 644                 wim_RO.pop("wim_password", None)
 645                 # TODO make a deep update of wim_port_mapping
 646                 if wim_RO:
 647                     await RO.edit("wim", RO_wim_id, descriptor=wim_RO)
 648
 649                 step = "Editing wim-account at RO tenant"
 650                 wim_account_RO = {}
 651                 if "config" in wim_content:
 652                     if "wim_port_mapping" in wim_content["config"]:
 653                         del wim_content["config"]["wim_port_mapping"]
 654                     if not wim_content["config"]:
 655                         del wim_content["config"]
 656                 if "wim_tenant_name" in wim_content:
 657                     wim_account_RO["wim_tenant_name"] = wim_content["wim_tenant_name"]
 658                 if "wim_password" in wim_content:
 659                     wim_account_RO["wim_password"] = wim_content["wim_password"]
 660                 if wim_content.get("wim_password"):
 661                     wim_account_RO["wim_password"] = self.db.decrypt(
 662                         wim_content["wim_password"],
 663                         schema_version=schema_version,
 664                         salt=wim_id,
 665                     )
 666                 if "config" in wim_content:
 667                     wim_account_RO["config"] = wim_content["config"]
 668                 if wim_content.get("config"):
 669                     for p in self.wim_config_encrypted:
 670                         if wim_content["config"].get(p):
 671                             wim_account_RO["config"][p] = self.db.decrypt(
 672                                 wim_content["config"][p],
 673                                 schema_version=schema_version,
 674                                 salt=wim_id,
 675                             )
 676
 677                 if "wim_user" in wim_content:
 678                     wim_content["wim_username"] = wim_content["wim_user"]
 679                 # wim_account must be edited always even if empty in order to ensure changes are translated to RO
 680                 # wim_thread. RO will remove and relaunch a new thread for this wim_account
 681                 await RO.edit("wim_account", RO_wim_id, descriptor=wim_account_RO)
 682                 db_wim_update["_admin.operationalState"] = "ENABLED"
 683                 # Mark the WIM 'edit' HA task as successful
 684             operation_state = "COMPLETED"
 685             operation_details = "Done"
 686
 687             self.logger.debug(logging_text + "Exit Ok RO_wim_id={}".format(RO_wim_id))
 688             return
 689
 690         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 691             self.logger.error(logging_text + "Exit Exception {}".format(e))
 692             exc = e
 693         except Exception as e:
 694             self.logger.critical(
 695                 logging_text + "Exit Exception {}".format(e), exc_info=True
 696             )
 697             exc = e
 698         finally:
 699             if exc and db_wim:
 700                 db_wim_update["_admin.operationalState"] = "ERROR"
 701                 db_wim_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 702                     step, exc
 703                 )
 704                 # Mark the WIM 'edit' HA task as erroneous
 705                 operation_state = "FAILED"
 706                 operation_details = "ERROR {}: {}".format(step, exc)
 707             try:
 708                 if db_wim_update:
 709                     self.update_db_2("wim_accounts", wim_id, db_wim_update)
 710                 # Register the WIM 'edit' HA task either
 711                 # succesful or erroneous, or do nothing (if legacy NBI)
 712                 self.lcm_tasks.unlock_HA(
 713                     "wim",
 714                     "edit",
 715                     op_id,
 716                     operationState=operation_state,
 717                     detailed_status=operation_details,
 718                 )
 719             except DbException as e:
 720                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 721             self.lcm_tasks.remove("wim_account", wim_id, order_id)
 722
 723     async def delete(self, wim_content, order_id):
 724
 725         # HA tasks and backward compatibility:
 726         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
 727         # In such a case, HA is not supported by NBI, and the HA check always returns True
 728         op_id = wim_content.pop("op_id", None)
 729         if not self.lcm_tasks.lock_HA("wim", "delete", op_id):
 730             return
 731
 732         wim_id = wim_content["_id"]
 733         logging_text = "Task wim_delete={} ".format(wim_id)
 734         self.logger.debug(logging_text + "Enter")
 735
 736         db_wim = None
 737         db_wim_update = {}
 738         exc = None
 739         step = "Getting wim from db"
 740         try:
 741             # wait for any previous tasks in process
 742             await self.lcm_tasks.waitfor_related_HA("wim", "delete", op_id)
 743
 744             db_wim = self.db.get_one("wim_accounts", {"_id": wim_id})
 745             if (
 746                 db_wim.get("_admin")
 747                 and db_wim["_admin"].get("deployed")
 748                 and db_wim["_admin"]["deployed"].get("RO")
 749             ):
 750                 RO_wim_id = db_wim["_admin"]["deployed"]["RO"]
 751                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 752                 step = "Detaching wim from RO tenant"
 753                 try:
 754                     await RO.detach("wim_account", RO_wim_id)
 755                 except ROclient.ROClientException as e:
 756                     if e.http_code == 404:  # not found
 757                         self.logger.debug(
 758                             logging_text
 759                             + "RO_wim_id={} already detached".format(RO_wim_id)
 760                         )
 761                     else:
 762                         raise
 763
 764                 step = "Deleting wim from RO"
 765                 try:
 766                     await RO.delete("wim", RO_wim_id)
 767                 except ROclient.ROClientException as e:
 768                     if e.http_code == 404:  # not found
 769                         self.logger.debug(
 770                             logging_text
 771                             + "RO_wim_id={} already deleted".format(RO_wim_id)
 772                         )
 773                     else:
 774                         raise
 775             else:
 776                 # nothing to delete
 777                 self.logger.error(logging_text + "Nothing to remove at RO")
 778             self.db.del_one("wim_accounts", {"_id": wim_id})
 779             db_wim = None
 780             self.logger.debug(logging_text + "Exit Ok")
 781             return
 782
 783         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 784             self.logger.error(logging_text + "Exit Exception {}".format(e))
 785             exc = e
 786         except Exception as e:
 787             self.logger.critical(
 788                 logging_text + "Exit Exception {}".format(e), exc_info=True
 789             )
 790             exc = e
 791         finally:
 792             self.lcm_tasks.remove("wim_account", wim_id, order_id)
 793             if exc and db_wim:
 794                 db_wim_update["_admin.operationalState"] = "ERROR"
 795                 db_wim_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 796                     step, exc
 797                 )
 798                 # Mark the WIM 'delete' HA task as erroneous
 799                 operation_state = "FAILED"
 800                 operation_details = "ERROR {}: {}".format(step, exc)
 801                 self.lcm_tasks.unlock_HA(
 802                     "wim",
 803                     "delete",
 804                     op_id,
 805                     operationState=operation_state,
 806                     detailed_status=operation_details,
 807                 )
 808             try:
 809                 if db_wim and db_wim_update:
 810                     self.update_db_2("wim_accounts", wim_id, db_wim_update)
 811                 # If the WIM 'delete' HA task was succesful, the DB entry has been deleted,
 812                 # which means that there is nowhere to register this task, so do nothing here.
 813             except DbException as e:
 814                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 815             self.lcm_tasks.remove("wim_account", wim_id, order_id)
 816
 817
 818 class SdnLcm(LcmBase):
 819     def __init__(self, msg, lcm_tasks, config, loop):
 820         """
 821         Init, Connect to database, filesystem storage, and messaging
 822         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
 823         :return: None
 824         """
 825
 826         self.logger = logging.getLogger("lcm.sdn")
 827         self.loop = loop
 828         self.lcm_tasks = lcm_tasks
 829         self.ro_config = config["ro_config"]
 830
 831         super().__init__(msg, self.logger)
 832
 833     async def create(self, sdn_content, order_id):
 834
 835         # HA tasks and backward compatibility:
 836         # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
 837         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
 838         # Register 'create' task here for related future HA operations
 839         op_id = sdn_content.pop("op_id", None)
 840         self.lcm_tasks.lock_HA("sdn", "create", op_id)
 841
 842         sdn_id = sdn_content["_id"]
 843         logging_text = "Task sdn_create={} ".format(sdn_id)
 844         self.logger.debug(logging_text + "Enter")
 845
 846         db_sdn = None
 847         db_sdn_update = {}
 848         RO_sdn_id = None
 849         exc = None
 850         try:
 851             step = "Getting sdn from db"
 852             db_sdn = self.db.get_one("sdns", {"_id": sdn_id})
 853             db_sdn_update["_admin.deployed.RO"] = None
 854
 855             step = "Creating sdn at RO"
 856             db_sdn_update["_admin.detailed-status"] = step
 857             self.update_db_2("sdns", sdn_id, db_sdn_update)
 858
 859             RO = ROclient.ROClient(self.loop, **self.ro_config)
 860             sdn_RO = deepcopy(sdn_content)
 861             sdn_RO.pop("_id", None)
 862             sdn_RO.pop("_admin", None)
 863             schema_version = sdn_RO.pop("schema_version", None)
 864             sdn_RO.pop("schema_type", None)
 865             sdn_RO.pop("description", None)
 866             if sdn_RO.get("password"):
 867                 sdn_RO["password"] = self.db.decrypt(
 868                     sdn_RO["password"], schema_version=schema_version, salt=sdn_id
 869                 )
 870
 871             desc = await RO.create("sdn", descriptor=sdn_RO)
 872             RO_sdn_id = desc["uuid"]
 873             db_sdn_update["_admin.deployed.RO"] = RO_sdn_id
 874             db_sdn_update["_admin.operationalState"] = "ENABLED"
 875             self.logger.debug(logging_text + "Exit Ok RO_sdn_id={}".format(RO_sdn_id))
 876             # Mark the SDN 'create' HA task as successful
 877             operation_state = "COMPLETED"
 878             operation_details = "Done"
 879             return
 880
 881         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 882             self.logger.error(logging_text + "Exit Exception {}".format(e))
 883             exc = e
 884         except Exception as e:
 885             self.logger.critical(
 886                 logging_text + "Exit Exception {}".format(e), exc_info=True
 887             )
 888             exc = e
 889         finally:
 890             if exc and db_sdn:
 891                 db_sdn_update["_admin.operationalState"] = "ERROR"
 892                 db_sdn_update["_admin.detailed-status"] = "ERROR {}: {}".format(
 893                     step, exc
 894                 )
 895                 # Mark the SDN 'create' HA task as erroneous
 896                 operation_state = "FAILED"
 897                 operation_details = "ERROR {}: {}".format(step, exc)
 898             try:
 899                 if db_sdn and db_sdn_update:
 900                     self.update_db_2("sdns", sdn_id, db_sdn_update)
 901                 # Register the SDN 'create' HA task either
 902                 # succesful or erroneous, or do nothing (if legacy NBI)
 903                 self.lcm_tasks.unlock_HA(
 904                     "sdn",
 905                     "create",
 906                     op_id,
 907                     operationState=operation_state,
 908                     detailed_status=operation_details,
 909                 )
 910             except DbException as e:
 911                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 912             self.lcm_tasks.remove("sdn", sdn_id, order_id)
 913
 914     async def edit(self, sdn_content, order_id):
 915
 916         # HA tasks and backward compatibility:
 917         # If 'sdn_content' does not include 'op_id', we a running a legacy NBI version.
 918         # In such a case, HA is not supported by NBI, and the HA check always returns True
 919         op_id = sdn_content.pop("op_id", None)
 920         if not self.lcm_tasks.lock_HA("sdn", "edit", op_id):
 921             return
 922
 923         sdn_id = sdn_content["_id"]
 924         logging_text = "Task sdn_edit={} ".format(sdn_id)
 925         self.logger.debug(logging_text + "Enter")
 926
 927         db_sdn = None
 928         db_sdn_update = {}
 929         exc = None
 930         step = "Getting sdn from db"
 931         try:
 932             # wait for any previous tasks in process
 933             await self.lcm_tasks.waitfor_related_HA("sdn", "edit", op_id)
 934
 935             db_sdn = self.db.get_one("sdns", {"_id": sdn_id})
 936             RO_sdn_id = None
 937             if (
 938                 db_sdn.get("_admin")
 939                 and db_sdn["_admin"].get("deployed")
 940                 and db_sdn["_admin"]["deployed"].get("RO")
 941             ):
 942                 RO_sdn_id = db_sdn["_admin"]["deployed"]["RO"]
 943                 RO = ROclient.ROClient(self.loop, **self.ro_config)
 944                 step = "Editing sdn at RO"
 945                 sdn_RO = deepcopy(sdn_content)
 946                 sdn_RO.pop("_id", None)
 947                 sdn_RO.pop("_admin", None)
 948                 schema_version = sdn_RO.pop("schema_version", None)
 949                 sdn_RO.pop("schema_type", None)
 950                 sdn_RO.pop("description", None)
 951                 if sdn_RO.get("password"):
 952                     sdn_RO["password"] = self.db.decrypt(
 953                         sdn_RO["password"], schema_version=schema_version, salt=sdn_id
 954                     )
 955                 if sdn_RO:
 956                     await RO.edit("sdn", RO_sdn_id, descriptor=sdn_RO)
 957                 db_sdn_update["_admin.operationalState"] = "ENABLED"
 958                 # Mark the SDN 'edit' HA task as successful
 959             operation_state = "COMPLETED"
 960             operation_details = "Done"
 961
 962             self.logger.debug(logging_text + "Exit Ok RO_sdn_id={}".format(RO_sdn_id))
 963             return
 964
 965         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
 966             self.logger.error(logging_text + "Exit Exception {}".format(e))
 967             exc = e
 968         except Exception as e:
 969             self.logger.critical(
 970                 logging_text + "Exit Exception {}".format(e), exc_info=True
 971             )
 972             exc = e
 973         finally:
 974             if exc and db_sdn:
 975                 db_sdn["_admin.operationalState"] = "ERROR"
 976                 db_sdn["_admin.detailed-status"] = "ERROR {}: {}".format(step, exc)
 977                 # Mark the SDN 'edit' HA task as erroneous
 978                 operation_state = "FAILED"
 979                 operation_details = "ERROR {}: {}".format(step, exc)
 980             try:
 981                 if db_sdn_update:
 982                     self.update_db_2("sdns", sdn_id, db_sdn_update)
 983                 # Register the SDN 'edit' HA task either
 984                 # succesful or erroneous, or do nothing (if legacy NBI)
 985                 self.lcm_tasks.unlock_HA(
 986                     "sdn",
 987                     "edit",
 988                     op_id,
 989                     operationState=operation_state,
 990                     detailed_status=operation_details,
 991                 )
 992             except DbException as e:
 993                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
 994             self.lcm_tasks.remove("sdn", sdn_id, order_id)
 995
 996     async def delete(self, sdn_content, order_id):
 997
 998         # HA tasks and backward compatibility:
 999         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1000         # In such a case, HA is not supported by NBI, and the HA check always returns True
1001         op_id = sdn_content.pop("op_id", None)
1002         if not self.lcm_tasks.lock_HA("sdn", "delete", op_id):
1003             return
1004
1005         sdn_id = sdn_content["_id"]
1006         logging_text = "Task sdn_delete={} ".format(sdn_id)
1007         self.logger.debug(logging_text + "Enter")
1008
1009         db_sdn = None
1010         db_sdn_update = {}
1011         exc = None
1012         step = "Getting sdn from db"
1013         try:
1014             # wait for any previous tasks in process
1015             await self.lcm_tasks.waitfor_related_HA("sdn", "delete", op_id)
1016
1017             db_sdn = self.db.get_one("sdns", {"_id": sdn_id})
1018             if (
1019                 db_sdn.get("_admin")
1020                 and db_sdn["_admin"].get("deployed")
1021                 and db_sdn["_admin"]["deployed"].get("RO")
1022             ):
1023                 RO_sdn_id = db_sdn["_admin"]["deployed"]["RO"]
1024                 RO = ROclient.ROClient(self.loop, **self.ro_config)
1025                 step = "Deleting sdn from RO"
1026                 try:
1027                     await RO.delete("sdn", RO_sdn_id)
1028                 except ROclient.ROClientException as e:
1029                     if e.http_code == 404:  # not found
1030                         self.logger.debug(
1031                             logging_text
1032                             + "RO_sdn_id={} already deleted".format(RO_sdn_id)
1033                         )
1034                     else:
1035                         raise
1036             else:
1037                 # nothing to delete
1038                 self.logger.error(
1039                     logging_text + "Skipping. There is not RO information at database"
1040                 )
1041             self.db.del_one("sdns", {"_id": sdn_id})
1042             db_sdn = None
1043             self.logger.debug("sdn_delete task sdn_id={} Exit Ok".format(sdn_id))
1044             return
1045
1046         except (ROclient.ROClientException, DbException, asyncio.CancelledError) as e:
1047             self.logger.error(logging_text + "Exit Exception {}".format(e))
1048             exc = e
1049         except Exception as e:
1050             self.logger.critical(
1051                 logging_text + "Exit Exception {}".format(e), exc_info=True
1052             )
1053             exc = e
1054         finally:
1055             if exc and db_sdn:
1056                 db_sdn["_admin.operationalState"] = "ERROR"
1057                 db_sdn["_admin.detailed-status"] = "ERROR {}: {}".format(step, exc)
1058                 # Mark the SDN 'delete' HA task as erroneous
1059                 operation_state = "FAILED"
1060                 operation_details = "ERROR {}: {}".format(step, exc)
1061                 self.lcm_tasks.unlock_HA(
1062                     "sdn",
1063                     "delete",
1064                     op_id,
1065                     operationState=operation_state,
1066                     detailed_status=operation_details,
1067                 )
1068             try:
1069                 if db_sdn and db_sdn_update:
1070                     self.update_db_2("sdns", sdn_id, db_sdn_update)
1071                 # If the SDN 'delete' HA task was succesful, the DB entry has been deleted,
1072                 # which means that there is nowhere to register this task, so do nothing here.
1073             except DbException as e:
1074                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1075             self.lcm_tasks.remove("sdn", sdn_id, order_id)
1076
1077
1078 class K8sClusterLcm(LcmBase):
1079     timeout_create = 300
1080
1081     def __init__(self, msg, lcm_tasks, config, loop):
1082         """
1083         Init, Connect to database, filesystem storage, and messaging
1084         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1085         :return: None
1086         """
1087
1088         self.logger = logging.getLogger("lcm.k8scluster")
1089         self.loop = loop
1090         self.lcm_tasks = lcm_tasks
1091         self.vca_config = config["VCA"]
1092
1093         super().__init__(msg, self.logger)
1094
1095         self.helm2_k8scluster = K8sHelmConnector(
1096             kubectl_command=self.vca_config.get("kubectlpath"),
1097             helm_command=self.vca_config.get("helmpath"),
1098             log=self.logger,
1099             on_update_db=None,
1100             db=self.db,
1101             fs=self.fs,
1102         )
1103
1104         self.helm3_k8scluster = K8sHelm3Connector(
1105             kubectl_command=self.vca_config.get("kubectlpath"),
1106             helm_command=self.vca_config.get("helm3path"),
1107             fs=self.fs,
1108             log=self.logger,
1109             db=self.db,
1110             on_update_db=None,
1111         )
1112
1113         self.juju_k8scluster = K8sJujuConnector(
1114             kubectl_command=self.vca_config.get("kubectlpath"),
1115             juju_command=self.vca_config.get("jujupath"),
1116             log=self.logger,
1117             loop=self.loop,
1118             on_update_db=None,
1119             db=self.db,
1120             fs=self.fs,
1121         )
1122
1123         self.k8s_map = {
1124             "helm-chart": self.helm2_k8scluster,
1125             "helm-chart-v3": self.helm3_k8scluster,
1126             "juju-bundle": self.juju_k8scluster,
1127         }
1128
1129     async def create(self, k8scluster_content, order_id):
1130
1131         op_id = k8scluster_content.pop("op_id", None)
1132         if not self.lcm_tasks.lock_HA("k8scluster", "create", op_id):
1133             return
1134
1135         k8scluster_id = k8scluster_content["_id"]
1136         logging_text = "Task k8scluster_create={} ".format(k8scluster_id)
1137         self.logger.debug(logging_text + "Enter")
1138
1139         db_k8scluster = None
1140         db_k8scluster_update = {}
1141         exc = None
1142         try:
1143             step = "Getting k8scluster-id='{}' from db".format(k8scluster_id)
1144             self.logger.debug(logging_text + step)
1145             db_k8scluster = self.db.get_one("k8sclusters", {"_id": k8scluster_id})
1146             self.db.encrypt_decrypt_fields(
1147                 db_k8scluster.get("credentials"),
1148                 "decrypt",
1149                 ["password", "secret"],
1150                 schema_version=db_k8scluster["schema_version"],
1151                 salt=db_k8scluster["_id"],
1152             )
1153             k8s_credentials = yaml.safe_dump(db_k8scluster.get("credentials"))
1154             pending_tasks = []
1155             task2name = {}
1156             init_target = deep_get(db_k8scluster, ("_admin", "init"))
1157             step = "Launching k8scluster init tasks"
1158
1159             k8s_deploy_methods = db_k8scluster.get("deployment_methods", {})
1160             # for backwards compatibility and all-false case
1161             if not any(k8s_deploy_methods.values()):
1162                 k8s_deploy_methods = {"helm-chart": True, "juju-bundle": True, "helm-chart-v3": True}
1163             deploy_methods = tuple(filter(k8s_deploy_methods.get, k8s_deploy_methods))
1164
1165             for task_name in deploy_methods:
1166                 if init_target and task_name not in init_target:
1167                     continue
1168                 task = asyncio.ensure_future(
1169                     self.k8s_map[task_name].init_env(
1170                         k8s_credentials,
1171                         reuse_cluster_uuid=k8scluster_id,
1172                         vca_id=db_k8scluster.get("vca_id"),
1173                     )
1174                 )
1175                 pending_tasks.append(task)
1176                 task2name[task] = task_name
1177
1178             error_text_list = []
1179             tasks_name_ok = []
1180             reached_timeout = False
1181             now = time()
1182
1183             while pending_tasks:
1184                 _timeout = max(
1185                     1, self.timeout_create - (time() - now)
1186                 )  # ensure not negative with max
1187                 step = "Waiting for k8scluster init tasks"
1188                 done, pending_tasks = await asyncio.wait(
1189                     pending_tasks, timeout=_timeout, return_when=asyncio.FIRST_COMPLETED
1190                 )
1191                 if not done:
1192                     # timeout. Set timeout is reached and process pending as if they hase been finished
1193                     done = pending_tasks
1194                     pending_tasks = None
1195                     reached_timeout = True
1196                 for task in done:
1197                     task_name = task2name[task]
1198                     if reached_timeout:
1199                         exc = "Timeout"
1200                     elif task.cancelled():
1201                         exc = "Cancelled"
1202                     else:
1203                         exc = task.exception()
1204
1205                     if exc:
1206                         error_text_list.append(
1207                             "Failing init {}: {}".format(task_name, exc)
1208                         )
1209                         db_k8scluster_update[
1210                             "_admin.{}.error_msg".format(task_name)
1211                         ] = str(exc)
1212                         db_k8scluster_update["_admin.{}.id".format(task_name)] = None
1213                         db_k8scluster_update[
1214                             "_admin.{}.operationalState".format(task_name)
1215                         ] = "ERROR"
1216                         self.logger.error(
1217                             logging_text + "{} init fail: {}".format(task_name, exc),
1218                             exc_info=not isinstance(exc, (N2VCException, str)),
1219                         )
1220                     else:
1221                         k8s_id, uninstall_sw = task.result()
1222                         tasks_name_ok.append(task_name)
1223                         self.logger.debug(
1224                             logging_text
1225                             + "{} init success. id={} created={}".format(
1226                                 task_name, k8s_id, uninstall_sw
1227                             )
1228                         )
1229                         db_k8scluster_update[
1230                             "_admin.{}.error_msg".format(task_name)
1231                         ] = None
1232                         db_k8scluster_update["_admin.{}.id".format(task_name)] = k8s_id
1233                         db_k8scluster_update[
1234                             "_admin.{}.created".format(task_name)
1235                         ] = uninstall_sw
1236                         db_k8scluster_update[
1237                             "_admin.{}.operationalState".format(task_name)
1238                         ] = "ENABLED"
1239                 # update database
1240                 step = "Updating database for " + task_name
1241                 self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
1242             if tasks_name_ok:
1243                 operation_details = "ready for " + ", ".join(tasks_name_ok)
1244                 operation_state = "COMPLETED"
1245                 db_k8scluster_update["_admin.operationalState"] = (
1246                     "ENABLED" if not error_text_list else "DEGRADED"
1247                 )
1248                 operation_details += "; " + ";".join(error_text_list)
1249             else:
1250                 db_k8scluster_update["_admin.operationalState"] = "ERROR"
1251                 operation_state = "FAILED"
1252                 operation_details = ";".join(error_text_list)
1253             db_k8scluster_update["_admin.detailed-status"] = operation_details
1254             self.logger.debug(logging_text + "Done. Result: " + operation_state)
1255             exc = None
1256
1257         except Exception as e:
1258             if isinstance(
1259                 e,
1260                 (
1261                     LcmException,
1262                     DbException,
1263                     K8sException,
1264                     N2VCException,
1265                     asyncio.CancelledError,
1266                 ),
1267             ):
1268                 self.logger.error(logging_text + "Exit Exception {}".format(e))
1269             else:
1270                 self.logger.critical(
1271                     logging_text + "Exit Exception {}".format(e), exc_info=True
1272                 )
1273             exc = e
1274         finally:
1275             if exc and db_k8scluster:
1276                 db_k8scluster_update["_admin.operationalState"] = "ERROR"
1277                 db_k8scluster_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1278                     step, exc
1279                 )
1280                 operation_state = "FAILED"
1281                 operation_details = "ERROR {}: {}".format(step, exc)
1282             try:
1283                 if db_k8scluster and db_k8scluster_update:
1284                     self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
1285
1286                 # Register the operation and unlock
1287                 self.lcm_tasks.unlock_HA(
1288                     "k8scluster",
1289                     "create",
1290                     op_id,
1291                     operationState=operation_state,
1292                     detailed_status=operation_details,
1293                 )
1294             except DbException as e:
1295                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1296             self.lcm_tasks.remove("k8scluster", k8scluster_id, order_id)
1297
1298     async def edit(self, k8scluster_content, order_id):
1299
1300         op_id = k8scluster_content.pop("op_id", None)
1301         if not self.lcm_tasks.lock_HA("k8scluster", "edit", op_id):
1302             return
1303
1304         k8scluster_id = k8scluster_content["_id"]
1305         logging_text = "Task k8scluster_edit={} ".format(k8scluster_id)
1306         self.logger.debug(logging_text + "Enter")
1307
1308         # TODO the implementation is pending and will be part of a new feature
1309         # It will support rotation of certificates, update of credentials and K8S API endpoint
1310         # At the moment the operation is set as completed
1311
1312         operation_state = "COMPLETED"
1313         operation_details = "Not implemented"
1314
1315         self.lcm_tasks.unlock_HA(
1316             "k8scluster",
1317             "edit",
1318             op_id,
1319             operationState=operation_state,
1320             detailed_status=operation_details,
1321         )
1322         self.lcm_tasks.remove("k8scluster", k8scluster_id, order_id)
1323
1324     async def delete(self, k8scluster_content, order_id):
1325
1326         # HA tasks and backward compatibility:
1327         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1328         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1329         # Register 'delete' task here for related future HA operations
1330         op_id = k8scluster_content.pop("op_id", None)
1331         if not self.lcm_tasks.lock_HA("k8scluster", "delete", op_id):
1332             return
1333
1334         k8scluster_id = k8scluster_content["_id"]
1335         logging_text = "Task k8scluster_delete={} ".format(k8scluster_id)
1336         self.logger.debug(logging_text + "Enter")
1337
1338         db_k8scluster = None
1339         db_k8scluster_update = {}
1340         exc = None
1341         try:
1342             step = "Getting k8scluster='{}' from db".format(k8scluster_id)
1343             self.logger.debug(logging_text + step)
1344             db_k8scluster = self.db.get_one("k8sclusters", {"_id": k8scluster_id})
1345             k8s_hc_id = deep_get(db_k8scluster, ("_admin", "helm-chart", "id"))
1346             k8s_h3c_id = deep_get(db_k8scluster, ("_admin", "helm-chart-v3", "id"))
1347             k8s_jb_id = deep_get(db_k8scluster, ("_admin", "juju-bundle", "id"))
1348
1349             cluster_removed = True
1350             if k8s_jb_id:  # delete in reverse order of creation
1351                 step = "Removing juju-bundle '{}'".format(k8s_jb_id)
1352                 uninstall_sw = (
1353                     deep_get(db_k8scluster, ("_admin", "juju-bundle", "created"))
1354                     or False
1355                 )
1356                 cluster_removed = await self.juju_k8scluster.reset(
1357                     cluster_uuid=k8s_jb_id,
1358                     uninstall_sw=uninstall_sw,
1359                     vca_id=db_k8scluster.get("vca_id"),
1360                 )
1361                 db_k8scluster_update["_admin.juju-bundle.id"] = None
1362                 db_k8scluster_update["_admin.juju-bundle.operationalState"] = "DISABLED"
1363
1364             if k8s_hc_id:
1365                 step = "Removing helm-chart '{}'".format(k8s_hc_id)
1366                 uninstall_sw = (
1367                     deep_get(db_k8scluster, ("_admin", "helm-chart", "created"))
1368                     or False
1369                 )
1370                 cluster_removed = await self.helm2_k8scluster.reset(
1371                     cluster_uuid=k8s_hc_id, uninstall_sw=uninstall_sw
1372                 )
1373                 db_k8scluster_update["_admin.helm-chart.id"] = None
1374                 db_k8scluster_update["_admin.helm-chart.operationalState"] = "DISABLED"
1375
1376             if k8s_h3c_id:
1377                 step = "Removing helm-chart-v3 '{}'".format(k8s_hc_id)
1378                 uninstall_sw = (
1379                     deep_get(db_k8scluster, ("_admin", "helm-chart-v3", "created"))
1380                     or False
1381                 )
1382                 cluster_removed = await self.helm3_k8scluster.reset(
1383                     cluster_uuid=k8s_h3c_id, uninstall_sw=uninstall_sw
1384                 )
1385                 db_k8scluster_update["_admin.helm-chart-v3.id"] = None
1386                 db_k8scluster_update[
1387                     "_admin.helm-chart-v3.operationalState"
1388                 ] = "DISABLED"
1389
1390             # Try to remove from cluster_inserted to clean old versions
1391             if k8s_hc_id and cluster_removed:
1392                 step = "Removing k8scluster='{}' from k8srepos".format(k8scluster_id)
1393                 self.logger.debug(logging_text + step)
1394                 db_k8srepo_list = self.db.get_list(
1395                     "k8srepos", {"_admin.cluster-inserted": k8s_hc_id}
1396                 )
1397                 for k8srepo in db_k8srepo_list:
1398                     try:
1399                         cluster_list = k8srepo["_admin"]["cluster-inserted"]
1400                         cluster_list.remove(k8s_hc_id)
1401                         self.update_db_2(
1402                             "k8srepos",
1403                             k8srepo["_id"],
1404                             {"_admin.cluster-inserted": cluster_list},
1405                         )
1406                     except Exception as e:
1407                         self.logger.error("{}: {}".format(step, e))
1408             self.db.del_one("k8sclusters", {"_id": k8scluster_id})
1409             db_k8scluster_update = None
1410             self.logger.debug(logging_text + "Done")
1411
1412         except Exception as e:
1413             if isinstance(
1414                 e,
1415                 (
1416                     LcmException,
1417                     DbException,
1418                     K8sException,
1419                     N2VCException,
1420                     asyncio.CancelledError,
1421                 ),
1422             ):
1423                 self.logger.error(logging_text + "Exit Exception {}".format(e))
1424             else:
1425                 self.logger.critical(
1426                     logging_text + "Exit Exception {}".format(e), exc_info=True
1427                 )
1428             exc = e
1429         finally:
1430             if exc and db_k8scluster:
1431                 db_k8scluster_update["_admin.operationalState"] = "ERROR"
1432                 db_k8scluster_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1433                     step, exc
1434                 )
1435                 # Mark the WIM 'create' HA task as erroneous
1436                 operation_state = "FAILED"
1437                 operation_details = "ERROR {}: {}".format(step, exc)
1438             else:
1439                 operation_state = "COMPLETED"
1440                 operation_details = "deleted"
1441
1442             try:
1443                 if db_k8scluster_update:
1444                     self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
1445                 # Register the K8scluster 'delete' HA task either
1446                 # succesful or erroneous, or do nothing (if legacy NBI)
1447                 self.lcm_tasks.unlock_HA(
1448                     "k8scluster",
1449                     "delete",
1450                     op_id,
1451                     operationState=operation_state,
1452                     detailed_status=operation_details,
1453                 )
1454             except DbException as e:
1455                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1456             self.lcm_tasks.remove("k8scluster", k8scluster_id, order_id)
1457
1458
1459 class VcaLcm(LcmBase):
1460     timeout_create = 30
1461
1462     def __init__(self, msg, lcm_tasks, config, loop):
1463         """
1464         Init, Connect to database, filesystem storage, and messaging
1465         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1466         :return: None
1467         """
1468
1469         self.logger = logging.getLogger("lcm.vca")
1470         self.loop = loop
1471         self.lcm_tasks = lcm_tasks
1472
1473         super().__init__(msg, self.logger)
1474
1475         # create N2VC connector
1476         self.n2vc = N2VCJujuConnector(
1477             log=self.logger, loop=self.loop, fs=self.fs, db=self.db
1478         )
1479
1480     def _get_vca_by_id(self, vca_id: str) -> dict:
1481         db_vca = self.db.get_one("vca", {"_id": vca_id})
1482         self.db.encrypt_decrypt_fields(
1483             db_vca,
1484             "decrypt",
1485             ["secret", "cacert"],
1486             schema_version=db_vca["schema_version"],
1487             salt=db_vca["_id"],
1488         )
1489         return db_vca
1490
1491     async def create(self, vca_content, order_id):
1492         op_id = vca_content.pop("op_id", None)
1493         if not self.lcm_tasks.lock_HA("vca", "create", op_id):
1494             return
1495
1496         vca_id = vca_content["_id"]
1497         self.logger.debug("Task vca_create={} {}".format(vca_id, "Enter"))
1498
1499         db_vca = None
1500         db_vca_update = {}
1501
1502         try:
1503             self.logger.debug(
1504                 "Task vca_create={} {}".format(vca_id, "Getting vca from db")
1505             )
1506             db_vca = self._get_vca_by_id(vca_id)
1507
1508             task = asyncio.ensure_future(
1509                 asyncio.wait_for(
1510                     self.n2vc.validate_vca(db_vca["_id"]),
1511                     timeout=self.timeout_create,
1512                 )
1513             )
1514
1515             await asyncio.wait([task], return_when=asyncio.FIRST_COMPLETED)
1516             if task.exception():
1517                 raise task.exception()
1518             self.logger.debug(
1519                 "Task vca_create={} {}".format(
1520                     vca_id, "vca registered and validated successfully"
1521                 )
1522             )
1523             db_vca_update["_admin.operationalState"] = "ENABLED"
1524             db_vca_update["_admin.detailed-status"] = "Connectivity: ok"
1525             operation_details = "VCA validated"
1526             operation_state = "COMPLETED"
1527
1528             self.logger.debug(
1529                 "Task vca_create={} {}".format(
1530                     vca_id, "Done. Result: {}".format(operation_state)
1531                 )
1532             )
1533
1534         except Exception as e:
1535             error_msg = "Failed with exception: {}".format(e)
1536             self.logger.error("Task vca_create={} {}".format(vca_id, error_msg))
1537             db_vca_update["_admin.operationalState"] = "ERROR"
1538             db_vca_update["_admin.detailed-status"] = error_msg
1539             operation_state = "FAILED"
1540             operation_details = error_msg
1541         finally:
1542             try:
1543                 self.update_db_2("vca", vca_id, db_vca_update)
1544
1545                 # Register the operation and unlock
1546                 self.lcm_tasks.unlock_HA(
1547                     "vca",
1548                     "create",
1549                     op_id,
1550                     operationState=operation_state,
1551                     detailed_status=operation_details,
1552                 )
1553             except DbException as e:
1554                 self.logger.error(
1555                     "Task vca_create={} {}".format(
1556                         vca_id, "Cannot update database: {}".format(e)
1557                     )
1558                 )
1559             self.lcm_tasks.remove("vca", vca_id, order_id)
1560
1561     async def delete(self, vca_content, order_id):
1562
1563         # HA tasks and backward compatibility:
1564         # If "vim_content" does not include "op_id", we a running a legacy NBI version.
1565         # In such a case, HA is not supported by NBI, "op_id" is None, and lock_HA() will do nothing.
1566         # Register "delete" task here for related future HA operations
1567         op_id = vca_content.pop("op_id", None)
1568         if not self.lcm_tasks.lock_HA("vca", "delete", op_id):
1569             return
1570
1571         db_vca_update = {}
1572         vca_id = vca_content["_id"]
1573
1574         try:
1575             self.logger.debug(
1576                 "Task vca_delete={} {}".format(vca_id, "Deleting vca from db")
1577             )
1578             self.db.del_one("vca", {"_id": vca_id})
1579             db_vca_update = None
1580             operation_details = "deleted"
1581             operation_state = "COMPLETED"
1582
1583             self.logger.debug(
1584                 "Task vca_delete={} {}".format(
1585                     vca_id, "Done. Result: {}".format(operation_state)
1586                 )
1587             )
1588         except Exception as e:
1589             error_msg = "Failed with exception: {}".format(e)
1590             self.logger.error("Task vca_delete={} {}".format(vca_id, error_msg))
1591             db_vca_update["_admin.operationalState"] = "ERROR"
1592             db_vca_update["_admin.detailed-status"] = error_msg
1593             operation_state = "FAILED"
1594             operation_details = error_msg
1595         finally:
1596             try:
1597                 self.update_db_2("vca", vca_id, db_vca_update)
1598                 self.lcm_tasks.unlock_HA(
1599                     "vca",
1600                     "delete",
1601                     op_id,
1602                     operationState=operation_state,
1603                     detailed_status=operation_details,
1604                 )
1605             except DbException as e:
1606                 self.logger.error(
1607                     "Task vca_delete={} {}".format(
1608                         vca_id, "Cannot update database: {}".format(e)
1609                     )
1610                 )
1611             self.lcm_tasks.remove("vca", vca_id, order_id)
1612
1613
1614 class K8sRepoLcm(LcmBase):
1615     def __init__(self, msg, lcm_tasks, config, loop):
1616         """
1617         Init, Connect to database, filesystem storage, and messaging
1618         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1619         :return: None
1620         """
1621
1622         self.logger = logging.getLogger("lcm.k8srepo")
1623         self.loop = loop
1624         self.lcm_tasks = lcm_tasks
1625         self.vca_config = config["VCA"]
1626
1627         super().__init__(msg, self.logger)
1628
1629         self.k8srepo = K8sHelmConnector(
1630             kubectl_command=self.vca_config.get("kubectlpath"),
1631             helm_command=self.vca_config.get("helmpath"),
1632             fs=self.fs,
1633             log=self.logger,
1634             db=self.db,
1635             on_update_db=None,
1636         )
1637
1638     async def create(self, k8srepo_content, order_id):
1639
1640         # HA tasks and backward compatibility:
1641         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1642         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1643         # Register 'create' task here for related future HA operations
1644
1645         op_id = k8srepo_content.pop("op_id", None)
1646         if not self.lcm_tasks.lock_HA("k8srepo", "create", op_id):
1647             return
1648
1649         k8srepo_id = k8srepo_content.get("_id")
1650         logging_text = "Task k8srepo_create={} ".format(k8srepo_id)
1651         self.logger.debug(logging_text + "Enter")
1652
1653         db_k8srepo = None
1654         db_k8srepo_update = {}
1655         exc = None
1656         operation_state = "COMPLETED"
1657         operation_details = ""
1658         try:
1659             step = "Getting k8srepo-id='{}' from db".format(k8srepo_id)
1660             self.logger.debug(logging_text + step)
1661             db_k8srepo = self.db.get_one("k8srepos", {"_id": k8srepo_id})
1662             db_k8srepo_update["_admin.operationalState"] = "ENABLED"
1663         except Exception as e:
1664             self.logger.error(
1665                 logging_text + "Exit Exception {}".format(e),
1666                 exc_info=not isinstance(
1667                     e,
1668                     (
1669                         LcmException,
1670                         DbException,
1671                         K8sException,
1672                         N2VCException,
1673                         asyncio.CancelledError,
1674                     ),
1675                 ),
1676             )
1677             exc = e
1678         finally:
1679             if exc and db_k8srepo:
1680                 db_k8srepo_update["_admin.operationalState"] = "ERROR"
1681                 db_k8srepo_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1682                     step, exc
1683                 )
1684                 # Mark the WIM 'create' HA task as erroneous
1685                 operation_state = "FAILED"
1686                 operation_details = "ERROR {}: {}".format(step, exc)
1687             try:
1688                 if db_k8srepo_update:
1689                     self.update_db_2("k8srepos", k8srepo_id, db_k8srepo_update)
1690                 # Register the K8srepo 'create' HA task either
1691                 # succesful or erroneous, or do nothing (if legacy NBI)
1692                 self.lcm_tasks.unlock_HA(
1693                     "k8srepo",
1694                     "create",
1695                     op_id,
1696                     operationState=operation_state,
1697                     detailed_status=operation_details,
1698                 )
1699             except DbException as e:
1700                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1701             self.lcm_tasks.remove("k8srepo", k8srepo_id, order_id)
1702
1703     async def delete(self, k8srepo_content, order_id):
1704
1705         # HA tasks and backward compatibility:
1706         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1707         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1708         # Register 'delete' task here for related future HA operations
1709         op_id = k8srepo_content.pop("op_id", None)
1710         if not self.lcm_tasks.lock_HA("k8srepo", "delete", op_id):
1711             return
1712
1713         k8srepo_id = k8srepo_content.get("_id")
1714         logging_text = "Task k8srepo_delete={} ".format(k8srepo_id)
1715         self.logger.debug(logging_text + "Enter")
1716
1717         db_k8srepo = None
1718         db_k8srepo_update = {}
1719
1720         exc = None
1721         operation_state = "COMPLETED"
1722         operation_details = ""
1723         try:
1724             step = "Getting k8srepo-id='{}' from db".format(k8srepo_id)
1725             self.logger.debug(logging_text + step)
1726             db_k8srepo = self.db.get_one("k8srepos", {"_id": k8srepo_id})
1727
1728         except Exception as e:
1729             self.logger.error(
1730                 logging_text + "Exit Exception {}".format(e),
1731                 exc_info=not isinstance(
1732                     e,
1733                     (
1734                         LcmException,
1735                         DbException,
1736                         K8sException,
1737                         N2VCException,
1738                         asyncio.CancelledError,
1739                     ),
1740                 ),
1741             )
1742             exc = e
1743         finally:
1744             if exc and db_k8srepo:
1745                 db_k8srepo_update["_admin.operationalState"] = "ERROR"
1746                 db_k8srepo_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1747                     step, exc
1748                 )
1749                 # Mark the WIM 'create' HA task as erroneous
1750                 operation_state = "FAILED"
1751                 operation_details = "ERROR {}: {}".format(step, exc)
1752             try:
1753                 if db_k8srepo_update:
1754                     self.update_db_2("k8srepos", k8srepo_id, db_k8srepo_update)
1755                 # Register the K8srepo 'delete' HA task either
1756                 # succesful or erroneous, or do nothing (if legacy NBI)
1757                 self.lcm_tasks.unlock_HA(
1758                     "k8srepo",
1759                     "delete",
1760                     op_id,
1761                     operationState=operation_state,
1762                     detailed_status=operation_details,
1763                 )
1764                 self.db.del_one("k8srepos", {"_id": k8srepo_id})
1765             except DbException as e:
1766                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1767             self.lcm_tasks.remove("k8srepo", k8srepo_id, order_id)