n2vc/temporal_libjuju.py

   1 # Copyright 2023 Canonical Ltd.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at
   5 #
   6 #    http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  11 # implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 import logging
  16 from dataclasses import dataclass
  17 from typing import List
  18
  19 from juju.controller import Controller
  20 from juju.model import Model
  21
  22 from n2vc.exceptions import (
  23     JujuControllerFailedConnecting,
  24     JujuModelAlreadyExists,
  25 )
  26
  27
  28 @dataclass
  29 class ConnectionInfo:
  30     """Information to connect to juju controller"""
  31
  32     endpoint: str
  33     user: str
  34     password: str
  35     cacert: str
  36     cloud_name: str
  37     cloud_credentials: str
  38
  39     def __repr__(self):
  40         return f"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, caert: ******)"
  41
  42     def __str__(self):
  43         return f"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, caert: ******)"
  44
  45
  46 class Libjuju:
  47     def __init__(self, connection_info: ConnectionInfo) -> None:
  48         self.logger = logging.getLogger("temporal_libjuju")
  49         self.connection_info = connection_info
  50
  51     async def get_controller(self) -> Controller:
  52         controller = Controller()
  53         try:
  54             await controller.connect(
  55                 endpoint=self.connection_info.endpoint,
  56                 username=self.connection_info.user,
  57                 password=self.connection_info.password,
  58                 cacert=self.connection_info.cacert,
  59             )
  60             return controller
  61         except Exception as e:
  62             self.logger.error(
  63                 "Error connecting to controller={}: {}".format(
  64                     self.connection_info.endpoint, e
  65                 )
  66             )
  67             await self.disconnect_controller(controller)
  68             raise JujuControllerFailedConnecting(str(e))
  69
  70     async def disconnect_controller(self, controller: Controller) -> None:
  71         if controller:
  72             await controller.disconnect()
  73
  74     async def disconnect_model(self, model: Model):
  75         if model:
  76             await model.disconnect()
  77
  78     async def add_model(self, model_name: str):
  79         """Exception is raised if model_name already exists"""
  80         model = None
  81         controller = None
  82         try:
  83             controller = await self.get_controller()
  84             if await self.model_exists(model_name, controller=controller):
  85                 raise JujuModelAlreadyExists(
  86                     "Cannot create model {}".format(model_name)
  87                 )
  88             self.logger.debug("Creating model {}".format(model_name))
  89             model = await controller.add_model(
  90                 model_name,
  91                 cloud_name=self.connection_info.cloud_name,
  92                 credential_name=self.connection_info.cloud_credentials,
  93             )
  94         finally:
  95             await self.disconnect_model(model)
  96             await self.disconnect_controller(controller)
  97
  98     async def model_exists(
  99         self, model_name: str, controller: Controller = None
 100     ) -> bool:
 101         """Returns True if model exists. False otherwhise."""
 102         need_to_disconnect = False
 103         try:
 104             if not controller:
 105                 controller = await self.get_controller()
 106                 need_to_disconnect = True
 107
 108             return model_name in await controller.list_models()
 109         finally:
 110             if need_to_disconnect:
 111                 await self.disconnect_controller(controller)
 112
 113     async def get_model(self, controller: Controller, model_name: str) -> Model:
 114         return await controller.get_model(model_name)
 115
 116     async def list_models(self) -> List[str]:
 117         """List models in controller."""
 118         try:
 119             controller = await self.get_controller()
 120             return await controller.list_models()
 121         finally:
 122             await self.disconnect_controller(controller)
 123
 124     async def destroy_model(self, model_name: str, force=False) -> None:
 125         controller = None
 126         model = None
 127
 128         try:
 129             controller = await self.get_controller()
 130             if not await self.model_exists(model_name, controller=controller):
 131                 self.logger.warn(f"Model {model_name} doesn't exist")
 132                 return
 133
 134             self.logger.debug(f"Getting model {model_name} to destroy")
 135             model = await self.get_model(controller, model_name)
 136             await self.disconnect_model(model)
 137
 138             await controller.destroy_model(
 139                 model_name, destroy_storage=True, force=force, max_wait=60
 140             )
 141
 142         except Exception as e:
 143             self.logger.warn(f"Failed deleting model {model_name}: {e}")
 144             raise e
 145         finally:
 146             await self.disconnect_model(model)
 147             await self.disconnect_controller(controller)