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
6 # http://www.apache.org/licenses/LICENSE-2.0
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
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
16 from dataclasses
import dataclass
17 from typing
import List
19 from juju
.controller
import Controller
20 from juju
.errors
import JujuAPIError
, JujuError
21 from juju
.model
import Model
23 from n2vc
.exceptions
import JujuApplicationExists
, JujuControllerFailedConnecting
28 """Information to connect to juju controller"""
35 cloud_credentials
: str
38 return f
"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, caert: ******)"
41 return f
"{self.__class__.__name__}(endpoint: {self.endpoint}, user: {self.user}, password: ******, caert: ******)"
45 def __init__(self
, connection_info
: ConnectionInfo
) -> None:
46 self
.logger
= logging
.getLogger("temporal_libjuju")
47 self
.connection_info
= connection_info
49 async def get_controller(self
) -> Controller
:
50 controller
= Controller()
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
,
59 except Exception as e
:
61 "Error connecting to controller={}: {}".format(
62 self
.connection_info
.endpoint
, e
65 await self
.disconnect_controller(controller
)
66 raise JujuControllerFailedConnecting(str(e
))
68 async def disconnect_controller(self
, controller
: Controller
) -> None:
70 await controller
.disconnect()
72 async def disconnect_model(self
, model
: Model
):
74 await model
.disconnect()
76 async def add_model(self
, model_name
: str):
80 controller
= await self
.get_controller()
81 if await self
.model_exists(model_name
, controller
=controller
):
83 self
.logger
.debug("Creating model {}".format(model_name
))
84 model
= await controller
.add_model(
86 # config=self.vca_connection.data.model_config,
87 cloud_name
=self
.connection_info
.cloud_name
,
88 credential_name
=self
.connection_info
.cloud_credentials
,
90 except JujuAPIError
as e
:
91 if "already exists" in e
.message
:
96 await self
.disconnect_model(model
)
97 await self
.disconnect_controller(controller
)
99 async def model_exists(
100 self
, model_name
: str, controller
: Controller
= None
102 """Returns True if model exists. False otherwhise."""
103 need_to_disconnect
= False
106 controller
= await self
.get_controller()
107 need_to_disconnect
= True
109 return model_name
in await controller
.list_models()
111 if need_to_disconnect
:
112 await self
.disconnect_controller(controller
)
114 async def get_model(self
, model_name
: str, controller
: Controller
) -> Model
:
115 return await controller
.get_model(model_name
)
117 async def list_models(self
) -> List
[str]:
118 """List models in controller."""
120 controller
= await self
.get_controller()
121 return await controller
.list_models()
123 await self
.disconnect_controller(controller
)
125 async def deploy_charm(
127 application_name
: str,
133 channel
: str = "stable",
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.
146 (juju.application.Application): Juju application
149 "Deploying charm {} in model {}".format(application_name
, model_name
)
151 self
.logger
.debug("charm: {}".format(path
))
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
)
161 application
= await model
.deploy(
163 application_name
=application_name
,
171 "Wait until application {} is ready in model {}".format(
172 application_name
, model_name
175 await self
.wait_app_deployment_completion(application_name
, model_name
)
177 except JujuError
as e
:
178 if "already exists" in e
.message
:
179 raise JujuApplicationExists(
180 "Application {} exists".format(application_name
)
185 await self
.disconnect_model(model
)
186 await self
.disconnect_controller(controller
)
190 async def wait_app_deployment_completion(
191 self
, application_name
: str, model_name
: str
194 "Application {} is ready in model {}".format(application_name
, model_name
)
197 async def destroy_model(self
, model_name
: str, force
=False) -> None:
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")
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
)
211 await controller
.destroy_model(
212 model_name
, destroy_storage
=True, force
=force
, max_wait
=60
215 except Exception as e
:
216 self
.logger
.warn(f
"Failed deleting model {model_name}: {e}")
219 await self
.disconnect_model(model
)
220 await self
.disconnect_controller(controller
)