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