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 = {
1163                     "helm-chart": True,
1164                     "juju-bundle": True,
1165                     "helm-chart-v3": True,
1166                 }
1167             deploy_methods = tuple(filter(k8s_deploy_methods.get, k8s_deploy_methods))
1168
1169             for task_name in deploy_methods:
1170                 if init_target and task_name not in init_target:
1171                     continue
1172                 task = asyncio.ensure_future(
1173                     self.k8s_map[task_name].init_env(
1174                         k8s_credentials,
1175                         reuse_cluster_uuid=k8scluster_id,
1176                         vca_id=db_k8scluster.get("vca_id"),
1177                     )
1178                 )
1179                 pending_tasks.append(task)
1180                 task2name[task] = task_name
1181
1182             error_text_list = []
1183             tasks_name_ok = []
1184             reached_timeout = False
1185             now = time()
1186
1187             while pending_tasks:
1188                 _timeout = max(
1189                     1, self.timeout_create - (time() - now)
1190                 )  # ensure not negative with max
1191                 step = "Waiting for k8scluster init tasks"
1192                 done, pending_tasks = await asyncio.wait(
1193                     pending_tasks, timeout=_timeout, return_when=asyncio.FIRST_COMPLETED
1194                 )
1195                 if not done:
1196                     # timeout. Set timeout is reached and process pending as if they hase been finished
1197                     done = pending_tasks
1198                     pending_tasks = None
1199                     reached_timeout = True
1200                 for task in done:
1201                     task_name = task2name[task]
1202                     if reached_timeout:
1203                         exc = "Timeout"
1204                     elif task.cancelled():
1205                         exc = "Cancelled"
1206                     else:
1207                         exc = task.exception()
1208
1209                     if exc:
1210                         error_text_list.append(
1211                             "Failing init {}: {}".format(task_name, exc)
1212                         )
1213                         db_k8scluster_update[
1214                             "_admin.{}.error_msg".format(task_name)
1215                         ] = str(exc)
1216                         db_k8scluster_update["_admin.{}.id".format(task_name)] = None
1217                         db_k8scluster_update[
1218                             "_admin.{}.operationalState".format(task_name)
1219                         ] = "ERROR"
1220                         self.logger.error(
1221                             logging_text + "{} init fail: {}".format(task_name, exc),
1222                             exc_info=not isinstance(exc, (N2VCException, str)),
1223                         )
1224                     else:
1225                         k8s_id, uninstall_sw = task.result()
1226                         tasks_name_ok.append(task_name)
1227                         self.logger.debug(
1228                             logging_text
1229                             + "{} init success. id={} created={}".format(
1230                                 task_name, k8s_id, uninstall_sw
1231                             )
1232                         )
1233                         db_k8scluster_update[
1234                             "_admin.{}.error_msg".format(task_name)
1235                         ] = None
1236                         db_k8scluster_update["_admin.{}.id".format(task_name)] = k8s_id
1237                         db_k8scluster_update[
1238                             "_admin.{}.created".format(task_name)
1239                         ] = uninstall_sw
1240                         db_k8scluster_update[
1241                             "_admin.{}.operationalState".format(task_name)
1242                         ] = "ENABLED"
1243                 # update database
1244                 step = "Updating database for " + task_name
1245                 self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
1246             if tasks_name_ok:
1247                 operation_details = "ready for " + ", ".join(tasks_name_ok)
1248                 operation_state = "COMPLETED"
1249                 db_k8scluster_update["_admin.operationalState"] = (
1250                     "ENABLED" if not error_text_list else "DEGRADED"
1251                 )
1252                 operation_details += "; " + ";".join(error_text_list)
1253             else:
1254                 db_k8scluster_update["_admin.operationalState"] = "ERROR"
1255                 operation_state = "FAILED"
1256                 operation_details = ";".join(error_text_list)
1257             db_k8scluster_update["_admin.detailed-status"] = operation_details
1258             self.logger.debug(logging_text + "Done. Result: " + operation_state)
1259             exc = None
1260
1261         except Exception as e:
1262             if isinstance(
1263                 e,
1264                 (
1265                     LcmException,
1266                     DbException,
1267                     K8sException,
1268                     N2VCException,
1269                     asyncio.CancelledError,
1270                 ),
1271             ):
1272                 self.logger.error(logging_text + "Exit Exception {}".format(e))
1273             else:
1274                 self.logger.critical(
1275                     logging_text + "Exit Exception {}".format(e), exc_info=True
1276                 )
1277             exc = e
1278         finally:
1279             if exc and db_k8scluster:
1280                 db_k8scluster_update["_admin.operationalState"] = "ERROR"
1281                 db_k8scluster_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1282                     step, exc
1283                 )
1284                 operation_state = "FAILED"
1285                 operation_details = "ERROR {}: {}".format(step, exc)
1286             try:
1287                 if db_k8scluster and db_k8scluster_update:
1288                     self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
1289
1290                 # Register the operation and unlock
1291                 self.lcm_tasks.unlock_HA(
1292                     "k8scluster",
1293                     "create",
1294                     op_id,
1295                     operationState=operation_state,
1296                     detailed_status=operation_details,
1297                 )
1298             except DbException as e:
1299                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1300             self.lcm_tasks.remove("k8scluster", k8scluster_id, order_id)
1301
1302     async def delete(self, k8scluster_content, order_id):
1303
1304         # HA tasks and backward compatibility:
1305         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1306         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1307         # Register 'delete' task here for related future HA operations
1308         op_id = k8scluster_content.pop("op_id", None)
1309         if not self.lcm_tasks.lock_HA("k8scluster", "delete", op_id):
1310             return
1311
1312         k8scluster_id = k8scluster_content["_id"]
1313         logging_text = "Task k8scluster_delete={} ".format(k8scluster_id)
1314         self.logger.debug(logging_text + "Enter")
1315
1316         db_k8scluster = None
1317         db_k8scluster_update = {}
1318         exc = None
1319         try:
1320             step = "Getting k8scluster='{}' from db".format(k8scluster_id)
1321             self.logger.debug(logging_text + step)
1322             db_k8scluster = self.db.get_one("k8sclusters", {"_id": k8scluster_id})
1323             k8s_hc_id = deep_get(db_k8scluster, ("_admin", "helm-chart", "id"))
1324             k8s_h3c_id = deep_get(db_k8scluster, ("_admin", "helm-chart-v3", "id"))
1325             k8s_jb_id = deep_get(db_k8scluster, ("_admin", "juju-bundle", "id"))
1326
1327             cluster_removed = True
1328             if k8s_jb_id:  # delete in reverse order of creation
1329                 step = "Removing juju-bundle '{}'".format(k8s_jb_id)
1330                 uninstall_sw = (
1331                     deep_get(db_k8scluster, ("_admin", "juju-bundle", "created"))
1332                     or False
1333                 )
1334                 cluster_removed = await self.juju_k8scluster.reset(
1335                     cluster_uuid=k8s_jb_id,
1336                     uninstall_sw=uninstall_sw,
1337                     vca_id=db_k8scluster.get("vca_id"),
1338                 )
1339                 db_k8scluster_update["_admin.juju-bundle.id"] = None
1340                 db_k8scluster_update["_admin.juju-bundle.operationalState"] = "DISABLED"
1341
1342             if k8s_hc_id:
1343                 step = "Removing helm-chart '{}'".format(k8s_hc_id)
1344                 uninstall_sw = (
1345                     deep_get(db_k8scluster, ("_admin", "helm-chart", "created"))
1346                     or False
1347                 )
1348                 cluster_removed = await self.helm2_k8scluster.reset(
1349                     cluster_uuid=k8s_hc_id, uninstall_sw=uninstall_sw
1350                 )
1351                 db_k8scluster_update["_admin.helm-chart.id"] = None
1352                 db_k8scluster_update["_admin.helm-chart.operationalState"] = "DISABLED"
1353
1354             if k8s_h3c_id:
1355                 step = "Removing helm-chart-v3 '{}'".format(k8s_hc_id)
1356                 uninstall_sw = (
1357                     deep_get(db_k8scluster, ("_admin", "helm-chart-v3", "created"))
1358                     or False
1359                 )
1360                 cluster_removed = await self.helm3_k8scluster.reset(
1361                     cluster_uuid=k8s_h3c_id, uninstall_sw=uninstall_sw
1362                 )
1363                 db_k8scluster_update["_admin.helm-chart-v3.id"] = None
1364                 db_k8scluster_update[
1365                     "_admin.helm-chart-v3.operationalState"
1366                 ] = "DISABLED"
1367
1368             # Try to remove from cluster_inserted to clean old versions
1369             if k8s_hc_id and cluster_removed:
1370                 step = "Removing k8scluster='{}' from k8srepos".format(k8scluster_id)
1371                 self.logger.debug(logging_text + step)
1372                 db_k8srepo_list = self.db.get_list(
1373                     "k8srepos", {"_admin.cluster-inserted": k8s_hc_id}
1374                 )
1375                 for k8srepo in db_k8srepo_list:
1376                     try:
1377                         cluster_list = k8srepo["_admin"]["cluster-inserted"]
1378                         cluster_list.remove(k8s_hc_id)
1379                         self.update_db_2(
1380                             "k8srepos",
1381                             k8srepo["_id"],
1382                             {"_admin.cluster-inserted": cluster_list},
1383                         )
1384                     except Exception as e:
1385                         self.logger.error("{}: {}".format(step, e))
1386             self.db.del_one("k8sclusters", {"_id": k8scluster_id})
1387             db_k8scluster_update = None
1388             self.logger.debug(logging_text + "Done")
1389
1390         except Exception as e:
1391             if isinstance(
1392                 e,
1393                 (
1394                     LcmException,
1395                     DbException,
1396                     K8sException,
1397                     N2VCException,
1398                     asyncio.CancelledError,
1399                 ),
1400             ):
1401                 self.logger.error(logging_text + "Exit Exception {}".format(e))
1402             else:
1403                 self.logger.critical(
1404                     logging_text + "Exit Exception {}".format(e), exc_info=True
1405                 )
1406             exc = e
1407         finally:
1408             if exc and db_k8scluster:
1409                 db_k8scluster_update["_admin.operationalState"] = "ERROR"
1410                 db_k8scluster_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1411                     step, exc
1412                 )
1413                 # Mark the WIM 'create' HA task as erroneous
1414                 operation_state = "FAILED"
1415                 operation_details = "ERROR {}: {}".format(step, exc)
1416             else:
1417                 operation_state = "COMPLETED"
1418                 operation_details = "deleted"
1419
1420             try:
1421                 if db_k8scluster_update:
1422                     self.update_db_2("k8sclusters", k8scluster_id, db_k8scluster_update)
1423                 # Register the K8scluster 'delete' HA task either
1424                 # succesful or erroneous, or do nothing (if legacy NBI)
1425                 self.lcm_tasks.unlock_HA(
1426                     "k8scluster",
1427                     "delete",
1428                     op_id,
1429                     operationState=operation_state,
1430                     detailed_status=operation_details,
1431                 )
1432             except DbException as e:
1433                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1434             self.lcm_tasks.remove("k8scluster", k8scluster_id, order_id)
1435
1436
1437 class VcaLcm(LcmBase):
1438     timeout_create = 30
1439
1440     def __init__(self, msg, lcm_tasks, config, loop):
1441         """
1442         Init, Connect to database, filesystem storage, and messaging
1443         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1444         :return: None
1445         """
1446
1447         self.logger = logging.getLogger("lcm.vca")
1448         self.loop = loop
1449         self.lcm_tasks = lcm_tasks
1450
1451         super().__init__(msg, self.logger)
1452
1453         # create N2VC connector
1454         self.n2vc = N2VCJujuConnector(
1455             log=self.logger, loop=self.loop, fs=self.fs, db=self.db
1456         )
1457
1458     def _get_vca_by_id(self, vca_id: str) -> dict:
1459         db_vca = self.db.get_one("vca", {"_id": vca_id})
1460         self.db.encrypt_decrypt_fields(
1461             db_vca,
1462             "decrypt",
1463             ["secret", "cacert"],
1464             schema_version=db_vca["schema_version"],
1465             salt=db_vca["_id"],
1466         )
1467         return db_vca
1468
1469     async def create(self, vca_content, order_id):
1470         op_id = vca_content.pop("op_id", None)
1471         if not self.lcm_tasks.lock_HA("vca", "create", op_id):
1472             return
1473
1474         vca_id = vca_content["_id"]
1475         self.logger.debug("Task vca_create={} {}".format(vca_id, "Enter"))
1476
1477         db_vca = None
1478         db_vca_update = {}
1479
1480         try:
1481             self.logger.debug(
1482                 "Task vca_create={} {}".format(vca_id, "Getting vca from db")
1483             )
1484             db_vca = self._get_vca_by_id(vca_id)
1485
1486             task = asyncio.ensure_future(
1487                 asyncio.wait_for(
1488                     self.n2vc.validate_vca(db_vca["_id"]),
1489                     timeout=self.timeout_create,
1490                 )
1491             )
1492
1493             await asyncio.wait([task], return_when=asyncio.FIRST_COMPLETED)
1494             if task.exception():
1495                 raise task.exception()
1496             self.logger.debug(
1497                 "Task vca_create={} {}".format(
1498                     vca_id, "vca registered and validated successfully"
1499                 )
1500             )
1501             db_vca_update["_admin.operationalState"] = "ENABLED"
1502             db_vca_update["_admin.detailed-status"] = "Connectivity: ok"
1503             operation_details = "VCA validated"
1504             operation_state = "COMPLETED"
1505
1506             self.logger.debug(
1507                 "Task vca_create={} {}".format(
1508                     vca_id, "Done. Result: {}".format(operation_state)
1509                 )
1510             )
1511
1512         except Exception as e:
1513             error_msg = "Failed with exception: {}".format(e)
1514             self.logger.error("Task vca_create={} {}".format(vca_id, error_msg))
1515             db_vca_update["_admin.operationalState"] = "ERROR"
1516             db_vca_update["_admin.detailed-status"] = error_msg
1517             operation_state = "FAILED"
1518             operation_details = error_msg
1519         finally:
1520             try:
1521                 self.update_db_2("vca", vca_id, db_vca_update)
1522
1523                 # Register the operation and unlock
1524                 self.lcm_tasks.unlock_HA(
1525                     "vca",
1526                     "create",
1527                     op_id,
1528                     operationState=operation_state,
1529                     detailed_status=operation_details,
1530                 )
1531             except DbException as e:
1532                 self.logger.error(
1533                     "Task vca_create={} {}".format(
1534                         vca_id, "Cannot update database: {}".format(e)
1535                     )
1536                 )
1537             self.lcm_tasks.remove("vca", vca_id, order_id)
1538
1539     async def delete(self, vca_content, order_id):
1540
1541         # HA tasks and backward compatibility:
1542         # If "vim_content" does not include "op_id", we a running a legacy NBI version.
1543         # In such a case, HA is not supported by NBI, "op_id" is None, and lock_HA() will do nothing.
1544         # Register "delete" task here for related future HA operations
1545         op_id = vca_content.pop("op_id", None)
1546         if not self.lcm_tasks.lock_HA("vca", "delete", op_id):
1547             return
1548
1549         db_vca_update = {}
1550         vca_id = vca_content["_id"]
1551
1552         try:
1553             self.logger.debug(
1554                 "Task vca_delete={} {}".format(vca_id, "Deleting vca from db")
1555             )
1556             self.db.del_one("vca", {"_id": vca_id})
1557             db_vca_update = None
1558             operation_details = "deleted"
1559             operation_state = "COMPLETED"
1560
1561             self.logger.debug(
1562                 "Task vca_delete={} {}".format(
1563                     vca_id, "Done. Result: {}".format(operation_state)
1564                 )
1565             )
1566         except Exception as e:
1567             error_msg = "Failed with exception: {}".format(e)
1568             self.logger.error("Task vca_delete={} {}".format(vca_id, error_msg))
1569             db_vca_update["_admin.operationalState"] = "ERROR"
1570             db_vca_update["_admin.detailed-status"] = error_msg
1571             operation_state = "FAILED"
1572             operation_details = error_msg
1573         finally:
1574             try:
1575                 self.update_db_2("vca", vca_id, db_vca_update)
1576                 self.lcm_tasks.unlock_HA(
1577                     "vca",
1578                     "delete",
1579                     op_id,
1580                     operationState=operation_state,
1581                     detailed_status=operation_details,
1582                 )
1583             except DbException as e:
1584                 self.logger.error(
1585                     "Task vca_delete={} {}".format(
1586                         vca_id, "Cannot update database: {}".format(e)
1587                     )
1588                 )
1589             self.lcm_tasks.remove("vca", vca_id, order_id)
1590
1591
1592 class K8sRepoLcm(LcmBase):
1593     def __init__(self, msg, lcm_tasks, config, loop):
1594         """
1595         Init, Connect to database, filesystem storage, and messaging
1596         :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
1597         :return: None
1598         """
1599
1600         self.logger = logging.getLogger("lcm.k8srepo")
1601         self.loop = loop
1602         self.lcm_tasks = lcm_tasks
1603         self.vca_config = config["VCA"]
1604
1605         super().__init__(msg, self.logger)
1606
1607         self.k8srepo = K8sHelmConnector(
1608             kubectl_command=self.vca_config.get("kubectlpath"),
1609             helm_command=self.vca_config.get("helmpath"),
1610             fs=self.fs,
1611             log=self.logger,
1612             db=self.db,
1613             on_update_db=None,
1614         )
1615
1616     async def create(self, k8srepo_content, order_id):
1617
1618         # HA tasks and backward compatibility:
1619         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1620         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1621         # Register 'create' task here for related future HA operations
1622
1623         op_id = k8srepo_content.pop("op_id", None)
1624         if not self.lcm_tasks.lock_HA("k8srepo", "create", op_id):
1625             return
1626
1627         k8srepo_id = k8srepo_content.get("_id")
1628         logging_text = "Task k8srepo_create={} ".format(k8srepo_id)
1629         self.logger.debug(logging_text + "Enter")
1630
1631         db_k8srepo = None
1632         db_k8srepo_update = {}
1633         exc = None
1634         operation_state = "COMPLETED"
1635         operation_details = ""
1636         try:
1637             step = "Getting k8srepo-id='{}' from db".format(k8srepo_id)
1638             self.logger.debug(logging_text + step)
1639             db_k8srepo = self.db.get_one("k8srepos", {"_id": k8srepo_id})
1640             db_k8srepo_update["_admin.operationalState"] = "ENABLED"
1641         except Exception as e:
1642             self.logger.error(
1643                 logging_text + "Exit Exception {}".format(e),
1644                 exc_info=not isinstance(
1645                     e,
1646                     (
1647                         LcmException,
1648                         DbException,
1649                         K8sException,
1650                         N2VCException,
1651                         asyncio.CancelledError,
1652                     ),
1653                 ),
1654             )
1655             exc = e
1656         finally:
1657             if exc and db_k8srepo:
1658                 db_k8srepo_update["_admin.operationalState"] = "ERROR"
1659                 db_k8srepo_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1660                     step, exc
1661                 )
1662                 # Mark the WIM 'create' HA task as erroneous
1663                 operation_state = "FAILED"
1664                 operation_details = "ERROR {}: {}".format(step, exc)
1665             try:
1666                 if db_k8srepo_update:
1667                     self.update_db_2("k8srepos", k8srepo_id, db_k8srepo_update)
1668                 # Register the K8srepo 'create' HA task either
1669                 # succesful or erroneous, or do nothing (if legacy NBI)
1670                 self.lcm_tasks.unlock_HA(
1671                     "k8srepo",
1672                     "create",
1673                     op_id,
1674                     operationState=operation_state,
1675                     detailed_status=operation_details,
1676                 )
1677             except DbException as e:
1678                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1679             self.lcm_tasks.remove("k8srepo", k8srepo_id, order_id)
1680
1681     async def delete(self, k8srepo_content, order_id):
1682
1683         # HA tasks and backward compatibility:
1684         # If 'vim_content' does not include 'op_id', we a running a legacy NBI version.
1685         # In such a case, HA is not supported by NBI, 'op_id' is None, and lock_HA() will do nothing.
1686         # Register 'delete' task here for related future HA operations
1687         op_id = k8srepo_content.pop("op_id", None)
1688         if not self.lcm_tasks.lock_HA("k8srepo", "delete", op_id):
1689             return
1690
1691         k8srepo_id = k8srepo_content.get("_id")
1692         logging_text = "Task k8srepo_delete={} ".format(k8srepo_id)
1693         self.logger.debug(logging_text + "Enter")
1694
1695         db_k8srepo = None
1696         db_k8srepo_update = {}
1697
1698         exc = None
1699         operation_state = "COMPLETED"
1700         operation_details = ""
1701         try:
1702             step = "Getting k8srepo-id='{}' from db".format(k8srepo_id)
1703             self.logger.debug(logging_text + step)
1704             db_k8srepo = self.db.get_one("k8srepos", {"_id": k8srepo_id})
1705
1706         except Exception as e:
1707             self.logger.error(
1708                 logging_text + "Exit Exception {}".format(e),
1709                 exc_info=not isinstance(
1710                     e,
1711                     (
1712                         LcmException,
1713                         DbException,
1714                         K8sException,
1715                         N2VCException,
1716                         asyncio.CancelledError,
1717                     ),
1718                 ),
1719             )
1720             exc = e
1721         finally:
1722             if exc and db_k8srepo:
1723                 db_k8srepo_update["_admin.operationalState"] = "ERROR"
1724                 db_k8srepo_update["_admin.detailed-status"] = "ERROR {}: {}".format(
1725                     step, exc
1726                 )
1727                 # Mark the WIM 'create' HA task as erroneous
1728                 operation_state = "FAILED"
1729                 operation_details = "ERROR {}: {}".format(step, exc)
1730             try:
1731                 if db_k8srepo_update:
1732                     self.update_db_2("k8srepos", k8srepo_id, db_k8srepo_update)
1733                 # Register the K8srepo 'delete' HA task either
1734                 # succesful or erroneous, or do nothing (if legacy NBI)
1735                 self.lcm_tasks.unlock_HA(
1736                     "k8srepo",
1737                     "delete",
1738                     op_id,
1739                     operationState=operation_state,
1740                     detailed_status=operation_details,
1741                 )
1742                 self.db.del_one("k8srepos", {"_id": k8srepo_id})
1743             except DbException as e:
1744                 self.logger.error(logging_text + "Cannot update database: {}".format(e))
1745             self.lcm_tasks.remove("k8srepo", k8srepo_id, order_id)