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.errors import JujuAPIError, JujuError
  21 from juju.model import Model
  22
  23 from n2vc.exceptions import JujuApplicationExists, JujuControllerFailedConnecting
  24
  25
  26 @dataclass
  27 class ConnectionInfo:
  28     """Information to connect to juju controller"""
  29
  30     endpoint: str
  31     user: str
  32     password: str
  33     cacert: str
  34     cloud_name: str
  35     cloud_credentials: str
  36
  37     def __repr__(self):
  38         return f"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, caert: ******)"
  39
  40     def __str__(self):
  41         return f"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, caert: ******)"
  42
  43
  44 class Libjuju:
  45     def __init__(self, connection_info: ConnectionInfo) -> None:
  46         self.logger = logging.getLogger("temporal_libjuju")
  47         self.connection_info = connection_info
  48
  49     async def get_controller(self) -> Controller:
  50         controller = Controller()
  51         try:
  52             await controller.connect(
  53                 endpoint=self.connection_info.endpoint,
  54                 username=self.connection_info.user,
  55                 password=self.connection_info.password,
  56                 cacert=self.connection_info.cacert,
  57             )
  58             return controller
  59         except Exception as e:
  60             self.logger.error(
  61                 "Error connecting to controller={}: {}".format(
  62                     self.connection_info.endpoint, e
  63                 )
  64             )
  65             await self.disconnect_controller(controller)
  66             raise JujuControllerFailedConnecting(str(e))
  67
  68     async def disconnect_controller(self, controller: Controller) -> None:
  69         if controller:
  70             await controller.disconnect()
  71
  72     async def disconnect_model(self, model: Model):
  73         if model:
  74             await model.disconnect()
  75
  76     async def add_model(self, model_name: str):
  77         model = None
  78         controller = None
  79         try:
  80             controller = await self.get_controller()
  81             if await self.model_exists(model_name, controller=controller):
  82                 return
  83             self.logger.debug("Creating model {}".format(model_name))
  84             model = await controller.add_model(
  85                 model_name,
  86                 # config=self.vca_connection.data.model_config,
  87                 cloud_name=self.connection_info.cloud_name,
  88                 credential_name=self.connection_info.cloud_credentials,
  89             )
  90         except JujuAPIError as e:
  91             if "already exists" in e.message:
  92                 pass
  93             else:
  94                 raise e
  95         finally:
  96             await self.disconnect_model(model)
  97             await self.disconnect_controller(controller)
  98
  99     async def model_exists(
 100         self, model_name: str, controller: Controller = None
 101     ) -> bool:
 102         """Returns True if model exists. False otherwhise."""
 103         need_to_disconnect = False
 104         try:
 105             if not controller:
 106                 controller = await self.get_controller()
 107                 need_to_disconnect = True
 108
 109             return model_name in await controller.list_models()
 110         finally:
 111             if need_to_disconnect:
 112                 await self.disconnect_controller(controller)
 113
 114     async def get_model(self, model_name: str, controller: Controller) -> Model:
 115         return await controller.get_model(model_name)
 116
 117     async def list_models(self) -> List[str]:
 118         """List models in controller."""
 119         try:
 120             controller = await self.get_controller()
 121             return await controller.list_models()
 122         finally:
 123             await self.disconnect_controller(controller)
 124
 125     async def deploy_charm(
 126         self,
 127         application_name: str,
 128         path: str,
 129         model_name: str,
 130         config: dict = None,
 131         series: str = None,
 132         num_units: int = 1,
 133         channel: str = "stable",
 134     ):
 135         """
 136         Args:
 137             application_name (str): Application name.
 138             path (str): Local path to the charm.
 139             model_name (str): Model name.
 140             config (dict): Config for the charm.
 141             series (str): Series of the charm.
 142             num_units (str): Number of units to deploy.
 143             channel (str): Charm store channel from which to retrieve the charm.
 144
 145         Returns:
 146             (juju.application.Application): Juju application
 147         """
 148         self.logger.debug(
 149             "Deploying charm {} in model {}".format(application_name, model_name)
 150         )
 151         self.logger.debug("charm: {}".format(path))
 152         controller = None
 153         model = None
 154         try:
 155             controller = await self.get_controller()
 156             model = await self.get_model(controller, model_name)
 157             if application_name in model.applications:
 158                 raise JujuApplicationExists(
 159                     "Application {} exists".format(application_name)
 160                 )
 161             application = await model.deploy(
 162                 entity_url=path,
 163                 application_name=application_name,
 164                 channel=channel,
 165                 num_units=num_units,
 166                 series=series,
 167                 config=config,
 168             )
 169
 170             self.logger.debug(
 171                 "Wait until application {} is ready in model {}".format(
 172                     application_name, model_name
 173                 )
 174             )
 175             await self.wait_app_deployment_completion(application_name, model_name)
 176
 177         except JujuError as e:
 178             if "already exists" in e.message:
 179                 raise JujuApplicationExists(
 180                     "Application {} exists".format(application_name)
 181                 )
 182             else:
 183                 raise e
 184         finally:
 185             await self.disconnect_model(model)
 186             await self.disconnect_controller(controller)
 187
 188         return application
 189
 190     async def wait_app_deployment_completion(
 191         self, application_name: str, model_name: str
 192     ) -> None:
 193         self.logger.debug(
 194             "Application {} is ready in model {}".format(application_name, model_name)
 195         )
 196
 197     async def destroy_model(self, model_name: str, force=False) -> None:
 198         controller = None
 199         model = None
 200
 201         try:
 202             controller = await self.get_controller()
 203             if not await self.model_exists(model_name, controller=controller):
 204                 self.logger.warn(f"Model {model_name} doesn't exist")
 205                 return
 206
 207             self.logger.debug(f"Getting model {model_name} to destroy")
 208             model = await self.get_model(controller, model_name)
 209             await self.disconnect_model(model)
 210
 211             await controller.destroy_model(
 212                 model_name, destroy_storage=True, force=force, max_wait=60
 213             )
 214
 215         except Exception as e:
 216             self.logger.warn(f"Failed deleting model {model_name}: {e}")
 217             raise e
 218         finally:
 219             await self.disconnect_model(model)
 220             await self.disconnect_controller(controller)