Wait for cloud-init to finish before provisioning
[osm/N2VC.git] / n2vc / libjuju.py
1 # Copyright 2020 Canonical Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14
15 import asyncio
16 import logging
17 import typing
18
19 import time
20
21 from juju.errors import JujuAPIError
22 from juju.model import Model
23 from juju.machine import Machine
24 from juju.application import Application
25 from juju.unit import Unit
26 from juju.client._definitions import (
27 FullStatus,
28 QueryApplicationOffersResults,
29 Cloud,
30 CloudCredential,
31 )
32 from juju.controller import Controller
33 from juju.client import client
34 from juju import tag
35
36 from n2vc.juju_watcher import JujuModelWatcher
37 from n2vc.provisioner import AsyncSSHProvisioner
38 from n2vc.n2vc_conn import N2VCConnector
39 from n2vc.exceptions import (
40 JujuMachineNotFound,
41 JujuApplicationNotFound,
42 JujuLeaderUnitNotFound,
43 JujuActionNotFound,
44 JujuControllerFailedConnecting,
45 JujuApplicationExists,
46 JujuInvalidK8sConfiguration,
47 JujuError,
48 )
49 from n2vc.vca.cloud import Cloud as VcaCloud
50 from n2vc.vca.connection import Connection
51 from kubernetes.client.configuration import Configuration
52 from retrying_async import retry
53
54
55 RBAC_LABEL_KEY_NAME = "rbac-id"
56
57
58 class Libjuju:
59 def __init__(
60 self,
61 vca_connection: Connection,
62 loop: asyncio.AbstractEventLoop = None,
63 log: logging.Logger = None,
64 n2vc: N2VCConnector = None,
65 ):
66 """
67 Constructor
68
69 :param: vca_connection: n2vc.vca.connection object
70 :param: loop: Asyncio loop
71 :param: log: Logger
72 :param: n2vc: N2VC object
73 """
74
75 self.log = log or logging.getLogger("Libjuju")
76 self.n2vc = n2vc
77 self.vca_connection = vca_connection
78
79 self.loop = loop or asyncio.get_event_loop()
80 self.loop.set_exception_handler(self.handle_exception)
81 self.creating_model = asyncio.Lock(loop=self.loop)
82
83 if self.vca_connection.is_default:
84 self.health_check_task = self._create_health_check_task()
85
86 def _create_health_check_task(self):
87 return self.loop.create_task(self.health_check())
88
89 async def get_controller(self, timeout: float = 60.0) -> Controller:
90 """
91 Get controller
92
93 :param: timeout: Time in seconds to wait for controller to connect
94 """
95 controller = None
96 try:
97 controller = Controller(loop=self.loop)
98 await asyncio.wait_for(
99 controller.connect(
100 endpoint=self.vca_connection.data.endpoints,
101 username=self.vca_connection.data.user,
102 password=self.vca_connection.data.secret,
103 cacert=self.vca_connection.data.cacert,
104 ),
105 timeout=timeout,
106 )
107 if self.vca_connection.is_default:
108 endpoints = await controller.api_endpoints
109 if not all(
110 endpoint in self.vca_connection.endpoints for endpoint in endpoints
111 ):
112 await self.vca_connection.update_endpoints(endpoints)
113 return controller
114 except asyncio.CancelledError as e:
115 raise e
116 except Exception as e:
117 self.log.error(
118 "Failed connecting to controller: {}... {}".format(
119 self.vca_connection.data.endpoints, e
120 )
121 )
122 if controller:
123 await self.disconnect_controller(controller)
124 raise JujuControllerFailedConnecting(e)
125
126 async def disconnect(self):
127 """Disconnect"""
128 # Cancel health check task
129 self.health_check_task.cancel()
130 self.log.debug("Libjuju disconnected!")
131
132 async def disconnect_model(self, model: Model):
133 """
134 Disconnect model
135
136 :param: model: Model that will be disconnected
137 """
138 await model.disconnect()
139
140 async def disconnect_controller(self, controller: Controller):
141 """
142 Disconnect controller
143
144 :param: controller: Controller that will be disconnected
145 """
146 if controller:
147 await controller.disconnect()
148
149 @retry(attempts=3, delay=5, timeout=None)
150 async def add_model(self, model_name: str, cloud: VcaCloud):
151 """
152 Create model
153
154 :param: model_name: Model name
155 :param: cloud: Cloud object
156 """
157
158 # Get controller
159 controller = await self.get_controller()
160 model = None
161 try:
162 # Block until other workers have finished model creation
163 while self.creating_model.locked():
164 await asyncio.sleep(0.1)
165
166 # Create the model
167 async with self.creating_model:
168 if await self.model_exists(model_name, controller=controller):
169 return
170 self.log.debug("Creating model {}".format(model_name))
171 model = await controller.add_model(
172 model_name,
173 config=self.vca_connection.data.model_config,
174 cloud_name=cloud.name,
175 credential_name=cloud.credential_name,
176 )
177 except JujuAPIError as e:
178 if "already exists" in e.message:
179 pass
180 else:
181 raise e
182 finally:
183 if model:
184 await self.disconnect_model(model)
185 await self.disconnect_controller(controller)
186
187 async def get_executed_actions(self, model_name: str) -> list:
188 """
189 Get executed/history of actions for a model.
190
191 :param: model_name: Model name, str.
192 :return: List of executed actions for a model.
193 """
194 model = None
195 executed_actions = []
196 controller = await self.get_controller()
197 try:
198 model = await self.get_model(controller, model_name)
199 # Get all unique action names
200 actions = {}
201 for application in model.applications:
202 application_actions = await self.get_actions(application, model_name)
203 actions.update(application_actions)
204 # Get status of all actions
205 for application_action in actions:
206 app_action_status_list = await model.get_action_status(
207 name=application_action
208 )
209 for action_id, action_status in app_action_status_list.items():
210 executed_action = {
211 "id": action_id,
212 "action": application_action,
213 "status": action_status,
214 }
215 # Get action output by id
216 action_status = await model.get_action_output(executed_action["id"])
217 for k, v in action_status.items():
218 executed_action[k] = v
219 executed_actions.append(executed_action)
220 except Exception as e:
221 raise JujuError(
222 "Error in getting executed actions for model: {}. Error: {}".format(
223 model_name, str(e)
224 )
225 )
226 finally:
227 if model:
228 await self.disconnect_model(model)
229 await self.disconnect_controller(controller)
230 return executed_actions
231
232 async def get_application_configs(
233 self, model_name: str, application_name: str
234 ) -> dict:
235 """
236 Get available configs for an application.
237
238 :param: model_name: Model name, str.
239 :param: application_name: Application name, str.
240
241 :return: A dict which has key - action name, value - action description
242 """
243 model = None
244 application_configs = {}
245 controller = await self.get_controller()
246 try:
247 model = await self.get_model(controller, model_name)
248 application = self._get_application(
249 model, application_name=application_name
250 )
251 application_configs = await application.get_config()
252 except Exception as e:
253 raise JujuError(
254 "Error in getting configs for application: {} in model: {}. Error: {}".format(
255 application_name, model_name, str(e)
256 )
257 )
258 finally:
259 if model:
260 await self.disconnect_model(model)
261 await self.disconnect_controller(controller)
262 return application_configs
263
264 @retry(attempts=3, delay=5)
265 async def get_model(self, controller: Controller, model_name: str) -> Model:
266 """
267 Get model from controller
268
269 :param: controller: Controller
270 :param: model_name: Model name
271
272 :return: Model: The created Juju model object
273 """
274 return await controller.get_model(model_name)
275
276 async def model_exists(
277 self, model_name: str, controller: Controller = None
278 ) -> bool:
279 """
280 Check if model exists
281
282 :param: controller: Controller
283 :param: model_name: Model name
284
285 :return bool
286 """
287 need_to_disconnect = False
288
289 # Get controller if not passed
290 if not controller:
291 controller = await self.get_controller()
292 need_to_disconnect = True
293
294 # Check if model exists
295 try:
296 return model_name in await controller.list_models()
297 finally:
298 if need_to_disconnect:
299 await self.disconnect_controller(controller)
300
301 async def models_exist(self, model_names: [str]) -> (bool, list):
302 """
303 Check if models exists
304
305 :param: model_names: List of strings with model names
306
307 :return (bool, list[str]): (True if all models exists, List of model names that don't exist)
308 """
309 if not model_names:
310 raise Exception(
311 "model_names must be a non-empty array. Given value: {}".format(
312 model_names
313 )
314 )
315 non_existing_models = []
316 models = await self.list_models()
317 existing_models = list(set(models).intersection(model_names))
318 non_existing_models = list(set(model_names) - set(existing_models))
319
320 return (
321 len(non_existing_models) == 0,
322 non_existing_models,
323 )
324
325 async def get_model_status(self, model_name: str) -> FullStatus:
326 """
327 Get model status
328
329 :param: model_name: Model name
330
331 :return: Full status object
332 """
333 controller = await self.get_controller()
334 model = await self.get_model(controller, model_name)
335 try:
336 return await model.get_status()
337 finally:
338 await self.disconnect_model(model)
339 await self.disconnect_controller(controller)
340
341 async def create_machine(
342 self,
343 model_name: str,
344 machine_id: str = None,
345 db_dict: dict = None,
346 progress_timeout: float = None,
347 total_timeout: float = None,
348 series: str = "xenial",
349 wait: bool = True,
350 ) -> (Machine, bool):
351 """
352 Create machine
353
354 :param: model_name: Model name
355 :param: machine_id: Machine id
356 :param: db_dict: Dictionary with data of the DB to write the updates
357 :param: progress_timeout: Maximum time between two updates in the model
358 :param: total_timeout: Timeout for the entity to be active
359 :param: series: Series of the machine (xenial, bionic, focal, ...)
360 :param: wait: Wait until machine is ready
361
362 :return: (juju.machine.Machine, bool): Machine object and a boolean saying
363 if the machine is new or it already existed
364 """
365 new = False
366 machine = None
367
368 self.log.debug(
369 "Creating machine (id={}) in model: {}".format(machine_id, model_name)
370 )
371
372 # Get controller
373 controller = await self.get_controller()
374
375 # Get model
376 model = await self.get_model(controller, model_name)
377 try:
378 if machine_id is not None:
379 self.log.debug(
380 "Searching machine (id={}) in model {}".format(
381 machine_id, model_name
382 )
383 )
384
385 # Get machines from model and get the machine with machine_id if exists
386 machines = await model.get_machines()
387 if machine_id in machines:
388 self.log.debug(
389 "Machine (id={}) found in model {}".format(
390 machine_id, model_name
391 )
392 )
393 machine = machines[machine_id]
394 else:
395 raise JujuMachineNotFound("Machine {} not found".format(machine_id))
396
397 if machine is None:
398 self.log.debug("Creating a new machine in model {}".format(model_name))
399
400 # Create machine
401 machine = await model.add_machine(
402 spec=None, constraints=None, disks=None, series=series
403 )
404 new = True
405
406 # Wait until the machine is ready
407 self.log.debug(
408 "Wait until machine {} is ready in model {}".format(
409 machine.entity_id, model_name
410 )
411 )
412 if wait:
413 await JujuModelWatcher.wait_for(
414 model=model,
415 entity=machine,
416 progress_timeout=progress_timeout,
417 total_timeout=total_timeout,
418 db_dict=db_dict,
419 n2vc=self.n2vc,
420 vca_id=self.vca_connection._vca_id,
421 )
422 finally:
423 await self.disconnect_model(model)
424 await self.disconnect_controller(controller)
425
426 self.log.debug(
427 "Machine {} ready at {} in model {}".format(
428 machine.entity_id, machine.dns_name, model_name
429 )
430 )
431 return machine, new
432
433 async def provision_machine(
434 self,
435 model_name: str,
436 hostname: str,
437 username: str,
438 private_key_path: str,
439 db_dict: dict = None,
440 progress_timeout: float = None,
441 total_timeout: float = None,
442 ) -> str:
443 """
444 Manually provisioning of a machine
445
446 :param: model_name: Model name
447 :param: hostname: IP to access the machine
448 :param: username: Username to login to the machine
449 :param: private_key_path: Local path for the private key
450 :param: db_dict: Dictionary with data of the DB to write the updates
451 :param: progress_timeout: Maximum time between two updates in the model
452 :param: total_timeout: Timeout for the entity to be active
453
454 :return: (Entity): Machine id
455 """
456 self.log.debug(
457 "Provisioning machine. model: {}, hostname: {}, username: {}".format(
458 model_name, hostname, username
459 )
460 )
461
462 # Get controller
463 controller = await self.get_controller()
464
465 # Get model
466 model = await self.get_model(controller, model_name)
467
468 try:
469 # Get provisioner
470 provisioner = AsyncSSHProvisioner(
471 host=hostname,
472 user=username,
473 private_key_path=private_key_path,
474 log=self.log,
475 )
476
477 # Provision machine
478 params = await provisioner.provision_machine()
479
480 params.jobs = ["JobHostUnits"]
481
482 self.log.debug("Adding machine to model")
483 connection = model.connection()
484 client_facade = client.ClientFacade.from_connection(connection)
485
486 results = await client_facade.AddMachines(params=[params])
487 error = results.machines[0].error
488
489 if error:
490 msg = "Error adding machine: {}".format(error.message)
491 self.log.error(msg=msg)
492 raise ValueError(msg)
493
494 machine_id = results.machines[0].machine
495
496 self.log.debug("Installing Juju agent into machine {}".format(machine_id))
497 asyncio.ensure_future(
498 provisioner.install_agent(
499 connection=connection,
500 nonce=params.nonce,
501 machine_id=machine_id,
502 proxy=self.vca_connection.data.api_proxy,
503 series=params.series,
504 )
505 )
506
507 machine = None
508 for _ in range(10):
509 machine_list = await model.get_machines()
510 if machine_id in machine_list:
511 self.log.debug("Machine {} found in model!".format(machine_id))
512 machine = model.machines.get(machine_id)
513 break
514 await asyncio.sleep(2)
515
516 if machine is None:
517 msg = "Machine {} not found in model".format(machine_id)
518 self.log.error(msg=msg)
519 raise JujuMachineNotFound(msg)
520
521 self.log.debug(
522 "Wait until machine {} is ready in model {}".format(
523 machine.entity_id, model_name
524 )
525 )
526 await JujuModelWatcher.wait_for(
527 model=model,
528 entity=machine,
529 progress_timeout=progress_timeout,
530 total_timeout=total_timeout,
531 db_dict=db_dict,
532 n2vc=self.n2vc,
533 vca_id=self.vca_connection._vca_id,
534 )
535 except Exception as e:
536 raise e
537 finally:
538 await self.disconnect_model(model)
539 await self.disconnect_controller(controller)
540
541 self.log.debug(
542 "Machine provisioned {} in model {}".format(machine_id, model_name)
543 )
544
545 return machine_id
546
547 async def deploy(
548 self, uri: str, model_name: str, wait: bool = True, timeout: float = 3600
549 ):
550 """
551 Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
552
553 :param: uri: Path or Charm Store uri in which the charm or bundle can be found
554 :param: model_name: Model name
555 :param: wait: Indicates whether to wait or not until all applications are active
556 :param: timeout: Time in seconds to wait until all applications are active
557 """
558 controller = await self.get_controller()
559 model = await self.get_model(controller, model_name)
560 try:
561 await model.deploy(uri)
562 if wait:
563 await JujuModelWatcher.wait_for_model(model, timeout=timeout)
564 self.log.debug("All units active in model {}".format(model_name))
565 finally:
566 await self.disconnect_model(model)
567 await self.disconnect_controller(controller)
568
569 async def deploy_charm(
570 self,
571 application_name: str,
572 path: str,
573 model_name: str,
574 machine_id: str,
575 db_dict: dict = None,
576 progress_timeout: float = None,
577 total_timeout: float = None,
578 config: dict = None,
579 series: str = None,
580 num_units: int = 1,
581 ):
582 """Deploy charm
583
584 :param: application_name: Application name
585 :param: path: Local path to the charm
586 :param: model_name: Model name
587 :param: machine_id ID of the machine
588 :param: db_dict: Dictionary with data of the DB to write the updates
589 :param: progress_timeout: Maximum time between two updates in the model
590 :param: total_timeout: Timeout for the entity to be active
591 :param: config: Config for the charm
592 :param: series: Series of the charm
593 :param: num_units: Number of units
594
595 :return: (juju.application.Application): Juju application
596 """
597 self.log.debug(
598 "Deploying charm {} to machine {} in model ~{}".format(
599 application_name, machine_id, model_name
600 )
601 )
602 self.log.debug("charm: {}".format(path))
603
604 # Get controller
605 controller = await self.get_controller()
606
607 # Get model
608 model = await self.get_model(controller, model_name)
609
610 try:
611 application = None
612 if application_name not in model.applications:
613
614 if machine_id is not None:
615 if machine_id not in model.machines:
616 msg = "Machine {} not found in model".format(machine_id)
617 self.log.error(msg=msg)
618 raise JujuMachineNotFound(msg)
619 machine = model.machines[machine_id]
620 series = machine.series
621
622 application = await model.deploy(
623 entity_url=path,
624 application_name=application_name,
625 channel="stable",
626 num_units=1,
627 series=series,
628 to=machine_id,
629 config=config,
630 )
631
632 self.log.debug(
633 "Wait until application {} is ready in model {}".format(
634 application_name, model_name
635 )
636 )
637 if num_units > 1:
638 for _ in range(num_units - 1):
639 m, _ = await self.create_machine(model_name, wait=False)
640 await application.add_unit(to=m.entity_id)
641
642 await JujuModelWatcher.wait_for(
643 model=model,
644 entity=application,
645 progress_timeout=progress_timeout,
646 total_timeout=total_timeout,
647 db_dict=db_dict,
648 n2vc=self.n2vc,
649 vca_id=self.vca_connection._vca_id,
650 )
651 self.log.debug(
652 "Application {} is ready in model {}".format(
653 application_name, model_name
654 )
655 )
656 else:
657 raise JujuApplicationExists(
658 "Application {} exists".format(application_name)
659 )
660 finally:
661 await self.disconnect_model(model)
662 await self.disconnect_controller(controller)
663
664 return application
665
666 async def scale_application(
667 self,
668 model_name: str,
669 application_name: str,
670 scale: int = 1,
671 total_timeout: float = None,
672 ):
673 """
674 Scale application (K8s)
675
676 :param: model_name: Model name
677 :param: application_name: Application name
678 :param: scale: Scale to which to set this application
679 :param: total_timeout: Timeout for the entity to be active
680 """
681
682 model = None
683 controller = await self.get_controller()
684 try:
685 model = await self.get_model(controller, model_name)
686
687 self.log.debug(
688 "Scaling application {} in model {}".format(
689 application_name, model_name
690 )
691 )
692 application = self._get_application(model, application_name)
693 if application is None:
694 raise JujuApplicationNotFound("Cannot scale application")
695 await application.scale(scale=scale)
696 # Wait until application is scaled in model
697 self.log.debug(
698 "Waiting for application {} to be scaled in model {}...".format(
699 application_name, model_name
700 )
701 )
702 if total_timeout is None:
703 total_timeout = 1800
704 end = time.time() + total_timeout
705 while time.time() < end:
706 application_scale = self._get_application_count(model, application_name)
707 # Before calling wait_for_model function,
708 # wait until application unit count and scale count are equal.
709 # Because there is a delay before scaling triggers in Juju model.
710 if application_scale == scale:
711 await JujuModelWatcher.wait_for_model(
712 model=model, timeout=total_timeout
713 )
714 self.log.debug(
715 "Application {} is scaled in model {}".format(
716 application_name, model_name
717 )
718 )
719 return
720 await asyncio.sleep(5)
721 raise Exception(
722 "Timeout waiting for application {} in model {} to be scaled".format(
723 application_name, model_name
724 )
725 )
726 finally:
727 if model:
728 await self.disconnect_model(model)
729 await self.disconnect_controller(controller)
730
731 def _get_application_count(self, model: Model, application_name: str) -> int:
732 """Get number of units of the application
733
734 :param: model: Model object
735 :param: application_name: Application name
736
737 :return: int (or None if application doesn't exist)
738 """
739 application = self._get_application(model, application_name)
740 if application is not None:
741 return len(application.units)
742
743 def _get_application(self, model: Model, application_name: str) -> Application:
744 """Get application
745
746 :param: model: Model object
747 :param: application_name: Application name
748
749 :return: juju.application.Application (or None if it doesn't exist)
750 """
751 if model.applications and application_name in model.applications:
752 return model.applications[application_name]
753
754 async def execute_action(
755 self,
756 application_name: str,
757 model_name: str,
758 action_name: str,
759 db_dict: dict = None,
760 progress_timeout: float = None,
761 total_timeout: float = None,
762 **kwargs
763 ):
764 """Execute action
765
766 :param: application_name: Application name
767 :param: model_name: Model name
768 :param: action_name: Name of the action
769 :param: db_dict: Dictionary with data of the DB to write the updates
770 :param: progress_timeout: Maximum time between two updates in the model
771 :param: total_timeout: Timeout for the entity to be active
772
773 :return: (str, str): (output and status)
774 """
775 self.log.debug(
776 "Executing action {} using params {}".format(action_name, kwargs)
777 )
778 # Get controller
779 controller = await self.get_controller()
780
781 # Get model
782 model = await self.get_model(controller, model_name)
783
784 try:
785 # Get application
786 application = self._get_application(
787 model,
788 application_name=application_name,
789 )
790 if application is None:
791 raise JujuApplicationNotFound("Cannot execute action")
792
793 # Get leader unit
794 # Racing condition:
795 # Ocassionally, self._get_leader_unit() will return None
796 # because the leader elected hook has not been triggered yet.
797 # Therefore, we are doing some retries. If it happens again,
798 # re-open bug 1236
799 unit = await self._get_leader_unit(application)
800
801 actions = await application.get_actions()
802
803 if action_name not in actions:
804 raise JujuActionNotFound(
805 "Action {} not in available actions".format(action_name)
806 )
807
808 action = await unit.run_action(action_name, **kwargs)
809
810 self.log.debug(
811 "Wait until action {} is completed in application {} (model={})".format(
812 action_name, application_name, model_name
813 )
814 )
815 await JujuModelWatcher.wait_for(
816 model=model,
817 entity=action,
818 progress_timeout=progress_timeout,
819 total_timeout=total_timeout,
820 db_dict=db_dict,
821 n2vc=self.n2vc,
822 vca_id=self.vca_connection._vca_id,
823 )
824
825 output = await model.get_action_output(action_uuid=action.entity_id)
826 status = await model.get_action_status(uuid_or_prefix=action.entity_id)
827 status = (
828 status[action.entity_id] if action.entity_id in status else "failed"
829 )
830
831 self.log.debug(
832 "Action {} completed with status {} in application {} (model={})".format(
833 action_name, action.status, application_name, model_name
834 )
835 )
836 finally:
837 await self.disconnect_model(model)
838 await self.disconnect_controller(controller)
839
840 return output, status
841
842 async def get_actions(self, application_name: str, model_name: str) -> dict:
843 """Get list of actions
844
845 :param: application_name: Application name
846 :param: model_name: Model name
847
848 :return: Dict with this format
849 {
850 "action_name": "Description of the action",
851 ...
852 }
853 """
854 self.log.debug(
855 "Getting list of actions for application {}".format(application_name)
856 )
857
858 # Get controller
859 controller = await self.get_controller()
860
861 # Get model
862 model = await self.get_model(controller, model_name)
863
864 try:
865 # Get application
866 application = self._get_application(
867 model,
868 application_name=application_name,
869 )
870
871 # Return list of actions
872 return await application.get_actions()
873
874 finally:
875 # Disconnect from model and controller
876 await self.disconnect_model(model)
877 await self.disconnect_controller(controller)
878
879 async def get_metrics(self, model_name: str, application_name: str) -> dict:
880 """Get the metrics collected by the VCA.
881
882 :param model_name The name or unique id of the network service
883 :param application_name The name of the application
884 """
885 if not model_name or not application_name:
886 raise Exception("model_name and application_name must be non-empty strings")
887 metrics = {}
888 controller = await self.get_controller()
889 model = await self.get_model(controller, model_name)
890 try:
891 application = self._get_application(model, application_name)
892 if application is not None:
893 metrics = await application.get_metrics()
894 finally:
895 self.disconnect_model(model)
896 self.disconnect_controller(controller)
897 return metrics
898
899 async def add_relation(
900 self,
901 model_name: str,
902 endpoint_1: str,
903 endpoint_2: str,
904 ):
905 """Add relation
906
907 :param: model_name: Model name
908 :param: endpoint_1 First endpoint name
909 ("app:endpoint" format or directly the saas name)
910 :param: endpoint_2: Second endpoint name (^ same format)
911 """
912
913 self.log.debug("Adding relation: {} -> {}".format(endpoint_1, endpoint_2))
914
915 # Get controller
916 controller = await self.get_controller()
917
918 # Get model
919 model = await self.get_model(controller, model_name)
920
921 # Add relation
922 try:
923 await model.add_relation(endpoint_1, endpoint_2)
924 except JujuAPIError as e:
925 if "not found" in e.message:
926 self.log.warning("Relation not found: {}".format(e.message))
927 return
928 if "already exists" in e.message:
929 self.log.warning("Relation already exists: {}".format(e.message))
930 return
931 # another exception, raise it
932 raise e
933 finally:
934 await self.disconnect_model(model)
935 await self.disconnect_controller(controller)
936
937 async def consume(
938 self,
939 offer_url: str,
940 model_name: str,
941 ):
942 """
943 Adds a remote offer to the model. Relations can be created later using "juju relate".
944
945 :param: offer_url: Offer Url
946 :param: model_name: Model name
947
948 :raises ParseError if there's a problem parsing the offer_url
949 :raises JujuError if remote offer includes and endpoint
950 :raises JujuAPIError if the operation is not successful
951 """
952 controller = await self.get_controller()
953 model = await controller.get_model(model_name)
954
955 try:
956 await model.consume(offer_url)
957 finally:
958 await self.disconnect_model(model)
959 await self.disconnect_controller(controller)
960
961 async def destroy_model(self, model_name: str, total_timeout: float):
962 """
963 Destroy model
964
965 :param: model_name: Model name
966 :param: total_timeout: Timeout
967 """
968
969 controller = await self.get_controller()
970 model = None
971 try:
972 if not await self.model_exists(model_name, controller=controller):
973 return
974
975 model = await self.get_model(controller, model_name)
976 self.log.debug("Destroying model {}".format(model_name))
977 uuid = model.info.uuid
978
979 # Destroy machines that are manually provisioned
980 # and still are in pending state
981 await self._destroy_pending_machines(model, only_manual=True)
982
983 # Disconnect model
984 await self.disconnect_model(model)
985
986 await controller.destroy_model(uuid, force=True, max_wait=0)
987
988 # Wait until model is destroyed
989 self.log.debug("Waiting for model {} to be destroyed...".format(model_name))
990
991 if total_timeout is None:
992 total_timeout = 3600
993 end = time.time() + total_timeout
994 while time.time() < end:
995 models = await controller.list_models()
996 if model_name not in models:
997 self.log.debug(
998 "The model {} ({}) was destroyed".format(model_name, uuid)
999 )
1000 return
1001 await asyncio.sleep(5)
1002 raise Exception(
1003 "Timeout waiting for model {} to be destroyed".format(model_name)
1004 )
1005 except Exception as e:
1006 if model:
1007 await self.disconnect_model(model)
1008 raise e
1009 finally:
1010 await self.disconnect_controller(controller)
1011
1012 async def destroy_application(
1013 self, model_name: str, application_name: str, total_timeout: float
1014 ):
1015 """
1016 Destroy application
1017
1018 :param: model_name: Model name
1019 :param: application_name: Application name
1020 :param: total_timeout: Timeout
1021 """
1022
1023 controller = await self.get_controller()
1024 model = None
1025
1026 try:
1027 model = await self.get_model(controller, model_name)
1028 self.log.debug(
1029 "Destroying application {} in model {}".format(
1030 application_name, model_name
1031 )
1032 )
1033 application = self._get_application(model, application_name)
1034 if application:
1035 await application.destroy()
1036 else:
1037 self.log.warning("Application not found: {}".format(application_name))
1038
1039 self.log.debug(
1040 "Waiting for application {} to be destroyed in model {}...".format(
1041 application_name, model_name
1042 )
1043 )
1044 if total_timeout is None:
1045 total_timeout = 3600
1046 end = time.time() + total_timeout
1047 while time.time() < end:
1048 if not self._get_application(model, application_name):
1049 self.log.debug(
1050 "The application {} was destroyed in model {} ".format(
1051 application_name, model_name
1052 )
1053 )
1054 return
1055 await asyncio.sleep(5)
1056 raise Exception(
1057 "Timeout waiting for application {} to be destroyed in model {}".format(
1058 application_name, model_name
1059 )
1060 )
1061 finally:
1062 if model is not None:
1063 await self.disconnect_model(model)
1064 await self.disconnect_controller(controller)
1065
1066 async def _destroy_pending_machines(self, model: Model, only_manual: bool = False):
1067 """
1068 Destroy pending machines in a given model
1069
1070 :param: only_manual: Bool that indicates only manually provisioned
1071 machines should be destroyed (if True), or that
1072 all pending machines should be destroyed
1073 """
1074 status = await model.get_status()
1075 for machine_id in status.machines:
1076 machine_status = status.machines[machine_id]
1077 if machine_status.agent_status.status == "pending":
1078 if only_manual and not machine_status.instance_id.startswith("manual:"):
1079 break
1080 machine = model.machines[machine_id]
1081 await machine.destroy(force=True)
1082
1083 async def configure_application(
1084 self, model_name: str, application_name: str, config: dict = None
1085 ):
1086 """Configure application
1087
1088 :param: model_name: Model name
1089 :param: application_name: Application name
1090 :param: config: Config to apply to the charm
1091 """
1092 self.log.debug("Configuring application {}".format(application_name))
1093
1094 if config:
1095 controller = await self.get_controller()
1096 model = None
1097 try:
1098 model = await self.get_model(controller, model_name)
1099 application = self._get_application(
1100 model,
1101 application_name=application_name,
1102 )
1103 await application.set_config(config)
1104 finally:
1105 if model:
1106 await self.disconnect_model(model)
1107 await self.disconnect_controller(controller)
1108
1109 def handle_exception(self, loop, context):
1110 # All unhandled exceptions by libjuju are handled here.
1111 pass
1112
1113 async def health_check(self, interval: float = 300.0):
1114 """
1115 Health check to make sure controller and controller_model connections are OK
1116
1117 :param: interval: Time in seconds between checks
1118 """
1119 controller = None
1120 while True:
1121 try:
1122 controller = await self.get_controller()
1123 # self.log.debug("VCA is alive")
1124 except Exception as e:
1125 self.log.error("Health check to VCA failed: {}".format(e))
1126 finally:
1127 await self.disconnect_controller(controller)
1128 await asyncio.sleep(interval)
1129
1130 async def list_models(self, contains: str = None) -> [str]:
1131 """List models with certain names
1132
1133 :param: contains: String that is contained in model name
1134
1135 :retur: [models] Returns list of model names
1136 """
1137
1138 controller = await self.get_controller()
1139 try:
1140 models = await controller.list_models()
1141 if contains:
1142 models = [model for model in models if contains in model]
1143 return models
1144 finally:
1145 await self.disconnect_controller(controller)
1146
1147 async def list_offers(self, model_name: str) -> QueryApplicationOffersResults:
1148 """List models with certain names
1149
1150 :param: model_name: Model name
1151
1152 :return: Returns list of offers
1153 """
1154
1155 controller = await self.get_controller()
1156 try:
1157 return await controller.list_offers(model_name)
1158 finally:
1159 await self.disconnect_controller(controller)
1160
1161 async def add_k8s(
1162 self,
1163 name: str,
1164 rbac_id: str,
1165 token: str,
1166 client_cert_data: str,
1167 configuration: Configuration,
1168 storage_class: str,
1169 credential_name: str = None,
1170 ):
1171 """
1172 Add a Kubernetes cloud to the controller
1173
1174 Similar to the `juju add-k8s` command in the CLI
1175
1176 :param: name: Name for the K8s cloud
1177 :param: configuration: Kubernetes configuration object
1178 :param: storage_class: Storage Class to use in the cloud
1179 :param: credential_name: Storage Class to use in the cloud
1180 """
1181
1182 if not storage_class:
1183 raise Exception("storage_class must be a non-empty string")
1184 if not name:
1185 raise Exception("name must be a non-empty string")
1186 if not configuration:
1187 raise Exception("configuration must be provided")
1188
1189 endpoint = configuration.host
1190 credential = self.get_k8s_cloud_credential(
1191 configuration,
1192 client_cert_data,
1193 token,
1194 )
1195 credential.attrs[RBAC_LABEL_KEY_NAME] = rbac_id
1196 cloud = client.Cloud(
1197 type_="kubernetes",
1198 auth_types=[credential.auth_type],
1199 endpoint=endpoint,
1200 ca_certificates=[client_cert_data],
1201 config={
1202 "operator-storage": storage_class,
1203 "workload-storage": storage_class,
1204 },
1205 )
1206
1207 return await self.add_cloud(
1208 name, cloud, credential, credential_name=credential_name
1209 )
1210
1211 def get_k8s_cloud_credential(
1212 self,
1213 configuration: Configuration,
1214 client_cert_data: str,
1215 token: str = None,
1216 ) -> client.CloudCredential:
1217 attrs = {}
1218 # TODO: Test with AKS
1219 key = None # open(configuration.key_file, "r").read()
1220 username = configuration.username
1221 password = configuration.password
1222
1223 if client_cert_data:
1224 attrs["ClientCertificateData"] = client_cert_data
1225 if key:
1226 attrs["ClientKeyData"] = key
1227 if token:
1228 if username or password:
1229 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1230 attrs["Token"] = token
1231
1232 auth_type = None
1233 if key:
1234 auth_type = "oauth2"
1235 if client_cert_data:
1236 auth_type = "oauth2withcert"
1237 if not token:
1238 raise JujuInvalidK8sConfiguration(
1239 "missing token for auth type {}".format(auth_type)
1240 )
1241 elif username:
1242 if not password:
1243 self.log.debug(
1244 "credential for user {} has empty password".format(username)
1245 )
1246 attrs["username"] = username
1247 attrs["password"] = password
1248 if client_cert_data:
1249 auth_type = "userpasswithcert"
1250 else:
1251 auth_type = "userpass"
1252 elif client_cert_data and token:
1253 auth_type = "certificate"
1254 else:
1255 raise JujuInvalidK8sConfiguration("authentication method not supported")
1256 return client.CloudCredential(auth_type=auth_type, attrs=attrs)
1257
1258 async def add_cloud(
1259 self,
1260 name: str,
1261 cloud: Cloud,
1262 credential: CloudCredential = None,
1263 credential_name: str = None,
1264 ) -> Cloud:
1265 """
1266 Add cloud to the controller
1267
1268 :param: name: Name of the cloud to be added
1269 :param: cloud: Cloud object
1270 :param: credential: CloudCredentials object for the cloud
1271 :param: credential_name: Credential name.
1272 If not defined, cloud of the name will be used.
1273 """
1274 controller = await self.get_controller()
1275 try:
1276 _ = await controller.add_cloud(name, cloud)
1277 if credential:
1278 await controller.add_credential(
1279 credential_name or name, credential=credential, cloud=name
1280 )
1281 # Need to return the object returned by the controller.add_cloud() function
1282 # I'm returning the original value now until this bug is fixed:
1283 # https://github.com/juju/python-libjuju/issues/443
1284 return cloud
1285 finally:
1286 await self.disconnect_controller(controller)
1287
1288 async def remove_cloud(self, name: str):
1289 """
1290 Remove cloud
1291
1292 :param: name: Name of the cloud to be removed
1293 """
1294 controller = await self.get_controller()
1295 try:
1296 await controller.remove_cloud(name)
1297 finally:
1298 await self.disconnect_controller(controller)
1299
1300 @retry(attempts=20, delay=5, fallback=JujuLeaderUnitNotFound())
1301 async def _get_leader_unit(self, application: Application) -> Unit:
1302 unit = None
1303 for u in application.units:
1304 if await u.is_leader_from_status():
1305 unit = u
1306 break
1307 if not unit:
1308 raise Exception()
1309 return unit
1310
1311 async def get_cloud_credentials(self, cloud: Cloud) -> typing.List:
1312 """
1313 Get cloud credentials
1314
1315 :param: cloud: Cloud object. The returned credentials will be from this cloud.
1316
1317 :return: List of credentials object associated to the specified cloud
1318
1319 """
1320 controller = await self.get_controller()
1321 try:
1322 facade = client.CloudFacade.from_connection(controller.connection())
1323 cloud_cred_tag = tag.credential(
1324 cloud.name, self.vca_connection.data.user, cloud.credential_name
1325 )
1326 params = [client.Entity(cloud_cred_tag)]
1327 return (await facade.Credential(params)).results
1328 finally:
1329 await self.disconnect_controller(controller)