1 # Copyright 2020 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
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 (
28 QueryApplicationOffersResults
,
32 from juju
.controller
import Controller
33 from juju
.client
import client
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 (
41 JujuApplicationNotFound
,
42 JujuLeaderUnitNotFound
,
44 JujuControllerFailedConnecting
,
45 JujuApplicationExists
,
46 JujuInvalidK8sConfiguration
,
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
55 RBAC_LABEL_KEY_NAME
= "rbac-id"
61 vca_connection
: Connection
,
62 loop
: asyncio
.AbstractEventLoop
= None,
63 log
: logging
.Logger
= None,
64 n2vc
: N2VCConnector
= None,
69 :param: vca_connection: n2vc.vca.connection object
70 :param: loop: Asyncio loop
72 :param: n2vc: N2VC object
75 self
.log
= log
or logging
.getLogger("Libjuju")
77 self
.vca_connection
= vca_connection
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
)
83 if self
.vca_connection
.is_default
:
84 self
.health_check_task
= self
._create
_health
_check
_task
()
86 def _create_health_check_task(self
):
87 return self
.loop
.create_task(self
.health_check())
89 async def get_controller(self
, timeout
: float = 60.0) -> Controller
:
93 :param: timeout: Time in seconds to wait for controller to connect
97 controller
= Controller()
98 await asyncio
.wait_for(
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
,
107 if self
.vca_connection
.is_default
:
108 endpoints
= await controller
.api_endpoints
110 endpoint
in self
.vca_connection
.endpoints
for endpoint
in endpoints
112 await self
.vca_connection
.update_endpoints(endpoints
)
114 except asyncio
.CancelledError
as e
:
116 except Exception as e
:
118 "Failed connecting to controller: {}... {}".format(
119 self
.vca_connection
.data
.endpoints
, e
123 await self
.disconnect_controller(controller
)
125 raise JujuControllerFailedConnecting(
126 f
"Error connecting to Juju controller: {e}"
129 async def disconnect(self
):
131 # Cancel health check task
132 self
.health_check_task
.cancel()
133 self
.log
.debug("Libjuju disconnected!")
135 async def disconnect_model(self
, model
: Model
):
139 :param: model: Model that will be disconnected
141 await model
.disconnect()
143 async def disconnect_controller(self
, controller
: Controller
):
145 Disconnect controller
147 :param: controller: Controller that will be disconnected
150 await controller
.disconnect()
152 @retry(attempts
=3, delay
=5, timeout
=None)
153 async def add_model(self
, model_name
: str, cloud
: VcaCloud
):
157 :param: model_name: Model name
158 :param: cloud: Cloud object
162 controller
= await self
.get_controller()
165 # Block until other workers have finished model creation
166 while self
.creating_model
.locked():
167 await asyncio
.sleep(0.1)
170 async with self
.creating_model
:
171 if await self
.model_exists(model_name
, controller
=controller
):
173 self
.log
.debug("Creating model {}".format(model_name
))
174 model
= await controller
.add_model(
176 config
=self
.vca_connection
.data
.model_config
,
177 cloud_name
=cloud
.name
,
178 credential_name
=cloud
.credential_name
,
180 except juju
.errors
.JujuAPIError
as e
:
181 if "already exists" in e
.message
:
187 await self
.disconnect_model(model
)
188 await self
.disconnect_controller(controller
)
190 async def get_executed_actions(self
, model_name
: str) -> list:
192 Get executed/history of actions for a model.
194 :param: model_name: Model name, str.
195 :return: List of executed actions for a model.
198 executed_actions
= []
199 controller
= await self
.get_controller()
201 model
= await self
.get_model(controller
, model_name
)
202 # Get all unique action names
204 for application
in model
.applications
:
205 application_actions
= await self
.get_actions(application
, model_name
)
206 actions
.update(application_actions
)
207 # Get status of all actions
208 for application_action
in actions
:
209 app_action_status_list
= await model
.get_action_status(
210 name
=application_action
212 for action_id
, action_status
in app_action_status_list
.items():
215 "action": application_action
,
216 "status": action_status
,
218 # Get action output by id
219 action_status
= await model
.get_action_output(executed_action
["id"])
220 for k
, v
in action_status
.items():
221 executed_action
[k
] = v
222 executed_actions
.append(executed_action
)
223 except Exception as e
:
225 "Error in getting executed actions for model: {}. Error: {}".format(
231 await self
.disconnect_model(model
)
232 await self
.disconnect_controller(controller
)
233 return executed_actions
235 async def get_application_configs(
236 self
, model_name
: str, application_name
: str
239 Get available configs for an application.
241 :param: model_name: Model name, str.
242 :param: application_name: Application name, str.
244 :return: A dict which has key - action name, value - action description
247 application_configs
= {}
248 controller
= await self
.get_controller()
250 model
= await self
.get_model(controller
, model_name
)
251 application
= self
._get
_application
(
252 model
, application_name
=application_name
254 application_configs
= await application
.get_config()
255 except Exception as e
:
257 "Error in getting configs for application: {} in model: {}. Error: {}".format(
258 application_name
, model_name
, str(e
)
263 await self
.disconnect_model(model
)
264 await self
.disconnect_controller(controller
)
265 return application_configs
267 @retry(attempts
=3, delay
=5)
268 async def get_model(self
, controller
: Controller
, model_name
: str) -> Model
:
270 Get model from controller
272 :param: controller: Controller
273 :param: model_name: Model name
275 :return: Model: The created Juju model object
277 return await controller
.get_model(model_name
)
279 async def model_exists(
280 self
, model_name
: str, controller
: Controller
= None
283 Check if model exists
285 :param: controller: Controller
286 :param: model_name: Model name
290 need_to_disconnect
= False
292 # Get controller if not passed
294 controller
= await self
.get_controller()
295 need_to_disconnect
= True
297 # Check if model exists
299 return model_name
in await controller
.list_models()
301 if need_to_disconnect
:
302 await self
.disconnect_controller(controller
)
304 async def models_exist(self
, model_names
: [str]) -> (bool, list):
306 Check if models exists
308 :param: model_names: List of strings with model names
310 :return (bool, list[str]): (True if all models exists, List of model names that don't exist)
314 "model_names must be a non-empty array. Given value: {}".format(
318 non_existing_models
= []
319 models
= await self
.list_models()
320 existing_models
= list(set(models
).intersection(model_names
))
321 non_existing_models
= list(set(model_names
) - set(existing_models
))
324 len(non_existing_models
) == 0,
328 async def get_model_status(self
, model_name
: str) -> FullStatus
:
332 :param: model_name: Model name
334 :return: Full status object
336 controller
= await self
.get_controller()
337 model
= await self
.get_model(controller
, model_name
)
339 return await model
.get_status()
341 await self
.disconnect_model(model
)
342 await self
.disconnect_controller(controller
)
344 async def create_machine(
347 machine_id
: str = None,
348 db_dict
: dict = None,
349 progress_timeout
: float = None,
350 total_timeout
: float = None,
351 series
: str = "bionic",
353 ) -> (Machine
, bool):
357 :param: model_name: Model name
358 :param: machine_id: Machine id
359 :param: db_dict: Dictionary with data of the DB to write the updates
360 :param: progress_timeout: Maximum time between two updates in the model
361 :param: total_timeout: Timeout for the entity to be active
362 :param: series: Series of the machine (xenial, bionic, focal, ...)
363 :param: wait: Wait until machine is ready
365 :return: (juju.machine.Machine, bool): Machine object and a boolean saying
366 if the machine is new or it already existed
372 "Creating machine (id={}) in model: {}".format(machine_id
, model_name
)
376 controller
= await self
.get_controller()
379 model
= await self
.get_model(controller
, model_name
)
381 if machine_id
is not None:
383 "Searching machine (id={}) in model {}".format(
384 machine_id
, model_name
388 # Get machines from model and get the machine with machine_id if exists
389 machines
= await model
.get_machines()
390 if machine_id
in machines
:
392 "Machine (id={}) found in model {}".format(
393 machine_id
, model_name
396 machine
= machines
[machine_id
]
398 raise JujuMachineNotFound("Machine {} not found".format(machine_id
))
401 self
.log
.debug("Creating a new machine in model {}".format(model_name
))
404 machine
= await model
.add_machine(
405 spec
=None, constraints
=None, disks
=None, series
=series
409 # Wait until the machine is ready
411 "Wait until machine {} is ready in model {}".format(
412 machine
.entity_id
, model_name
416 await JujuModelWatcher
.wait_for(
419 progress_timeout
=progress_timeout
,
420 total_timeout
=total_timeout
,
423 vca_id
=self
.vca_connection
._vca
_id
,
426 await self
.disconnect_model(model
)
427 await self
.disconnect_controller(controller
)
430 "Machine {} ready at {} in model {}".format(
431 machine
.entity_id
, machine
.dns_name
, model_name
436 async def provision_machine(
441 private_key_path
: str,
442 db_dict
: dict = None,
443 progress_timeout
: float = None,
444 total_timeout
: float = None,
447 Manually provisioning of a machine
449 :param: model_name: Model name
450 :param: hostname: IP to access the machine
451 :param: username: Username to login to the machine
452 :param: private_key_path: Local path for the private key
453 :param: db_dict: Dictionary with data of the DB to write the updates
454 :param: progress_timeout: Maximum time between two updates in the model
455 :param: total_timeout: Timeout for the entity to be active
457 :return: (Entity): Machine id
460 "Provisioning machine. model: {}, hostname: {}, username: {}".format(
461 model_name
, hostname
, username
466 controller
= await self
.get_controller()
469 model
= await self
.get_model(controller
, model_name
)
473 provisioner
= AsyncSSHProvisioner(
476 private_key_path
=private_key_path
,
481 params
= await provisioner
.provision_machine()
483 params
.jobs
= ["JobHostUnits"]
485 self
.log
.debug("Adding machine to model")
486 connection
= model
.connection()
487 client_facade
= client
.ClientFacade
.from_connection(connection
)
489 results
= await client_facade
.AddMachines(params
=[params
])
490 error
= results
.machines
[0].error
493 msg
= "Error adding machine: {}".format(error
.message
)
494 self
.log
.error(msg
=msg
)
495 raise ValueError(msg
)
497 machine_id
= results
.machines
[0].machine
499 self
.log
.debug("Installing Juju agent into machine {}".format(machine_id
))
500 asyncio
.ensure_future(
501 provisioner
.install_agent(
502 connection
=connection
,
504 machine_id
=machine_id
,
505 proxy
=self
.vca_connection
.data
.api_proxy
,
506 series
=params
.series
,
512 machine_list
= await model
.get_machines()
513 if machine_id
in machine_list
:
514 self
.log
.debug("Machine {} found in model!".format(machine_id
))
515 machine
= model
.machines
.get(machine_id
)
517 await asyncio
.sleep(2)
520 msg
= "Machine {} not found in model".format(machine_id
)
521 self
.log
.error(msg
=msg
)
522 raise JujuMachineNotFound(msg
)
525 "Wait until machine {} is ready in model {}".format(
526 machine
.entity_id
, model_name
529 await JujuModelWatcher
.wait_for(
532 progress_timeout
=progress_timeout
,
533 total_timeout
=total_timeout
,
536 vca_id
=self
.vca_connection
._vca
_id
,
538 except Exception as e
:
541 await self
.disconnect_model(model
)
542 await self
.disconnect_controller(controller
)
545 "Machine provisioned {} in model {}".format(machine_id
, model_name
)
551 self
, uri
: str, model_name
: str, wait
: bool = True, timeout
: float = 3600
554 Deploy bundle or charm: Similar to the juju CLI command `juju deploy`
556 :param: uri: Path or Charm Store uri in which the charm or bundle can be found
557 :param: model_name: Model name
558 :param: wait: Indicates whether to wait or not until all applications are active
559 :param: timeout: Time in seconds to wait until all applications are active
561 controller
= await self
.get_controller()
562 model
= await self
.get_model(controller
, model_name
)
564 await model
.deploy(uri
, trust
=True)
566 await JujuModelWatcher
.wait_for_model(model
, timeout
=timeout
)
567 self
.log
.debug("All units active in model {}".format(model_name
))
569 await self
.disconnect_model(model
)
570 await self
.disconnect_controller(controller
)
574 application_name
: str,
577 db_dict
: dict = None,
578 progress_timeout
: float = None,
579 total_timeout
: float = None,
583 :param: application_name: Application name
584 :param: model_name: Model name
585 :param: machine_id Machine id
586 :param: db_dict: Dictionary with data of the DB to write the updates
587 :param: progress_timeout: Maximum time between two updates in the model
588 :param: total_timeout: Timeout for the entity to be active
594 controller
= await self
.get_controller()
596 model
= await self
.get_model(controller
, model_name
)
597 application
= self
._get
_application
(model
, application_name
)
599 if application
is not None:
600 # Checks if the given machine id in the model,
601 # otherwise function raises an error
602 _machine
, _series
= self
._get
_machine
_info
(model
, machine_id
)
605 "Adding unit (machine {}) to application {} in model ~{}".format(
606 machine_id
, application_name
, model_name
610 await application
.add_unit(to
=machine_id
)
612 await JujuModelWatcher
.wait_for(
615 progress_timeout
=progress_timeout
,
616 total_timeout
=total_timeout
,
619 vca_id
=self
.vca_connection
._vca
_id
,
622 "Unit is added to application {} in model {}".format(
623 application_name
, model_name
627 raise JujuApplicationNotFound(
628 "Application {} not exists".format(application_name
)
632 await self
.disconnect_model(model
)
633 await self
.disconnect_controller(controller
)
635 async def destroy_unit(
637 application_name
: str,
640 total_timeout
: float = None,
644 :param: application_name: Application name
645 :param: model_name: Model name
646 :param: machine_id Machine id
647 :param: total_timeout: Timeout for the entity to be active
653 controller
= await self
.get_controller()
655 model
= await self
.get_model(controller
, model_name
)
656 application
= self
._get
_application
(model
, application_name
)
658 if application
is None:
659 raise JujuApplicationNotFound(
660 "Application not found: {} (model={})".format(
661 application_name
, model_name
665 unit
= self
._get
_unit
(application
, machine_id
)
668 "A unit with machine id {} not in available units".format(
673 unit_name
= unit
.name
676 "Destroying unit {} from application {} in model {}".format(
677 unit_name
, application_name
, model_name
680 await application
.destroy_unit(unit_name
)
683 "Waiting for unit {} to be destroyed in application {} (model={})...".format(
684 unit_name
, application_name
, model_name
688 # TODO: Add functionality in the Juju watcher to replace this kind of blocks
689 if total_timeout
is None:
691 end
= time
.time() + total_timeout
692 while time
.time() < end
:
693 if not self
._get
_unit
(application
, machine_id
):
695 "The unit {} was destroyed in application {} (model={}) ".format(
696 unit_name
, application_name
, model_name
700 await asyncio
.sleep(5)
702 "Unit {} is destroyed from application {} in model {}".format(
703 unit_name
, application_name
, model_name
708 await self
.disconnect_model(model
)
709 await self
.disconnect_controller(controller
)
711 async def deploy_charm(
713 application_name
: str,
717 db_dict
: dict = None,
718 progress_timeout
: float = None,
719 total_timeout
: float = None,
726 :param: application_name: Application name
727 :param: path: Local path to the charm
728 :param: model_name: Model name
729 :param: machine_id ID of the machine
730 :param: db_dict: Dictionary with data of the DB to write the updates
731 :param: progress_timeout: Maximum time between two updates in the model
732 :param: total_timeout: Timeout for the entity to be active
733 :param: config: Config for the charm
734 :param: series: Series of the charm
735 :param: num_units: Number of units
737 :return: (juju.application.Application): Juju application
740 "Deploying charm {} to machine {} in model ~{}".format(
741 application_name
, machine_id
, model_name
744 self
.log
.debug("charm: {}".format(path
))
747 controller
= await self
.get_controller()
750 model
= await self
.get_model(controller
, model_name
)
753 if application_name
not in model
.applications
:
754 if machine_id
is not None:
755 machine
, series
= self
._get
_machine
_info
(model
, machine_id
)
757 application
= await model
.deploy(
759 application_name
=application_name
,
768 "Wait until application {} is ready in model {}".format(
769 application_name
, model_name
773 for _
in range(num_units
- 1):
774 m
, _
= await self
.create_machine(model_name
, wait
=False)
775 await application
.add_unit(to
=m
.entity_id
)
777 await JujuModelWatcher
.wait_for(
780 progress_timeout
=progress_timeout
,
781 total_timeout
=total_timeout
,
784 vca_id
=self
.vca_connection
._vca
_id
,
787 "Application {} is ready in model {}".format(
788 application_name
, model_name
792 raise JujuApplicationExists(
793 "Application {} exists".format(application_name
)
795 except juju
.errors
.JujuError
as e
:
796 if "already exists" in e
.message
:
797 raise JujuApplicationExists(
798 "Application {} exists".format(application_name
)
803 await self
.disconnect_model(model
)
804 await self
.disconnect_controller(controller
)
808 async def resolve(self
, model_name
: str):
809 controller
= await self
.get_controller()
810 model
= await self
.get_model(controller
, model_name
)
811 all_units_active
= False
813 while not all_units_active
:
814 all_units_active
= True
815 for application_name
, application
in model
.applications
.items():
816 if application
.status
== "error":
817 for unit
in application
.units
:
818 if unit
.workload_status
== "error":
820 "Model {}, Application {}, Unit {} in error state, resolving".format(
821 model_name
, application_name
, unit
.entity_id
825 await unit
.resolved(retry
=False)
826 all_units_active
= False
830 if not all_units_active
:
831 await asyncio
.sleep(5)
833 await self
.disconnect_model(model
)
834 await self
.disconnect_controller(controller
)
836 async def scale_application(
839 application_name
: str,
841 total_timeout
: float = None,
844 Scale application (K8s)
846 :param: model_name: Model name
847 :param: application_name: Application name
848 :param: scale: Scale to which to set this application
849 :param: total_timeout: Timeout for the entity to be active
853 controller
= await self
.get_controller()
855 model
= await self
.get_model(controller
, model_name
)
858 "Scaling application {} in model {}".format(
859 application_name
, model_name
862 application
= self
._get
_application
(model
, application_name
)
863 if application
is None:
864 raise JujuApplicationNotFound("Cannot scale application")
865 await application
.scale(scale
=scale
)
866 # Wait until application is scaled in model
868 "Waiting for application {} to be scaled in model {}...".format(
869 application_name
, model_name
872 if total_timeout
is None:
874 end
= time
.time() + total_timeout
875 while time
.time() < end
:
876 application_scale
= self
._get
_application
_count
(model
, application_name
)
877 # Before calling wait_for_model function,
878 # wait until application unit count and scale count are equal.
879 # Because there is a delay before scaling triggers in Juju model.
880 if application_scale
== scale
:
881 await JujuModelWatcher
.wait_for_model(
882 model
=model
, timeout
=total_timeout
885 "Application {} is scaled in model {}".format(
886 application_name
, model_name
890 await asyncio
.sleep(5)
892 "Timeout waiting for application {} in model {} to be scaled".format(
893 application_name
, model_name
898 await self
.disconnect_model(model
)
899 await self
.disconnect_controller(controller
)
901 def _get_application_count(self
, model
: Model
, application_name
: str) -> int:
902 """Get number of units of the application
904 :param: model: Model object
905 :param: application_name: Application name
907 :return: int (or None if application doesn't exist)
909 application
= self
._get
_application
(model
, application_name
)
910 if application
is not None:
911 return len(application
.units
)
913 def _get_application(self
, model
: Model
, application_name
: str) -> Application
:
916 :param: model: Model object
917 :param: application_name: Application name
919 :return: juju.application.Application (or None if it doesn't exist)
921 if model
.applications
and application_name
in model
.applications
:
922 return model
.applications
[application_name
]
924 def _get_unit(self
, application
: Application
, machine_id
: str) -> Unit
:
927 :param: application: Application object
928 :param: machine_id: Machine id
933 for u
in application
.units
:
934 if u
.machine_id
== machine_id
:
939 def _get_machine_info(
946 :param: model: Model object
947 :param: machine_id: Machine id
949 :return: (str, str): (machine, series)
951 if machine_id
not in model
.machines
:
952 msg
= "Machine {} not found in model".format(machine_id
)
953 self
.log
.error(msg
=msg
)
954 raise JujuMachineNotFound(msg
)
955 machine
= model
.machines
[machine_id
]
956 return machine
, machine
.series
958 async def execute_action(
960 application_name
: str,
963 db_dict
: dict = None,
964 machine_id
: str = None,
965 progress_timeout
: float = None,
966 total_timeout
: float = None,
971 :param: application_name: Application name
972 :param: model_name: Model name
973 :param: action_name: Name of the action
974 :param: db_dict: Dictionary with data of the DB to write the updates
975 :param: machine_id Machine id
976 :param: progress_timeout: Maximum time between two updates in the model
977 :param: total_timeout: Timeout for the entity to be active
979 :return: (str, str): (output and status)
982 "Executing action {} using params {}".format(action_name
, kwargs
)
985 controller
= await self
.get_controller()
988 model
= await self
.get_model(controller
, model_name
)
992 application
= self
._get
_application
(
994 application_name
=application_name
,
996 if application
is None:
997 raise JujuApplicationNotFound("Cannot execute action")
999 # Ocassionally, self._get_leader_unit() will return None
1000 # because the leader elected hook has not been triggered yet.
1001 # Therefore, we are doing some retries. If it happens again,
1003 if machine_id
is None:
1004 unit
= await self
._get
_leader
_unit
(application
)
1006 "Action {} is being executed on the leader unit {}".format(
1007 action_name
, unit
.name
1011 unit
= self
._get
_unit
(application
, machine_id
)
1014 "A unit with machine id {} not in available units".format(
1019 "Action {} is being executed on {} unit".format(
1020 action_name
, unit
.name
1024 actions
= await application
.get_actions()
1026 if action_name
not in actions
:
1027 raise JujuActionNotFound(
1028 "Action {} not in available actions".format(action_name
)
1031 action
= await unit
.run_action(action_name
, **kwargs
)
1034 "Wait until action {} is completed in application {} (model={})".format(
1035 action_name
, application_name
, model_name
1038 await JujuModelWatcher
.wait_for(
1041 progress_timeout
=progress_timeout
,
1042 total_timeout
=total_timeout
,
1045 vca_id
=self
.vca_connection
._vca
_id
,
1048 output
= await model
.get_action_output(action_uuid
=action
.entity_id
)
1049 status
= await model
.get_action_status(uuid_or_prefix
=action
.entity_id
)
1051 status
[action
.entity_id
] if action
.entity_id
in status
else "failed"
1055 "Action {} completed with status {} in application {} (model={})".format(
1056 action_name
, action
.status
, application_name
, model_name
1060 await self
.disconnect_model(model
)
1061 await self
.disconnect_controller(controller
)
1063 return output
, status
1065 async def get_actions(self
, application_name
: str, model_name
: str) -> dict:
1066 """Get list of actions
1068 :param: application_name: Application name
1069 :param: model_name: Model name
1071 :return: Dict with this format
1073 "action_name": "Description of the action",
1078 "Getting list of actions for application {}".format(application_name
)
1082 controller
= await self
.get_controller()
1085 model
= await self
.get_model(controller
, model_name
)
1089 application
= self
._get
_application
(
1091 application_name
=application_name
,
1094 # Return list of actions
1095 return await application
.get_actions()
1098 # Disconnect from model and controller
1099 await self
.disconnect_model(model
)
1100 await self
.disconnect_controller(controller
)
1102 async def get_metrics(self
, model_name
: str, application_name
: str) -> dict:
1103 """Get the metrics collected by the VCA.
1105 :param model_name The name or unique id of the network service
1106 :param application_name The name of the application
1108 if not model_name
or not application_name
:
1109 raise Exception("model_name and application_name must be non-empty strings")
1111 controller
= await self
.get_controller()
1112 model
= await self
.get_model(controller
, model_name
)
1114 application
= self
._get
_application
(model
, application_name
)
1115 if application
is not None:
1116 metrics
= await application
.get_metrics()
1118 self
.disconnect_model(model
)
1119 self
.disconnect_controller(controller
)
1122 async def add_relation(
1130 :param: model_name: Model name
1131 :param: endpoint_1 First endpoint name
1132 ("app:endpoint" format or directly the saas name)
1133 :param: endpoint_2: Second endpoint name (^ same format)
1136 self
.log
.debug("Adding relation: {} -> {}".format(endpoint_1
, endpoint_2
))
1139 controller
= await self
.get_controller()
1142 model
= await self
.get_model(controller
, model_name
)
1146 await model
.add_relation(endpoint_1
, endpoint_2
)
1147 except juju
.errors
.JujuAPIError
as e
:
1148 if "not found" in e
.message
:
1149 self
.log
.warning("Relation not found: {}".format(e
.message
))
1151 if "already exists" in e
.message
:
1152 self
.log
.warning("Relation already exists: {}".format(e
.message
))
1154 # another exception, raise it
1157 await self
.disconnect_model(model
)
1158 await self
.disconnect_controller(controller
)
1166 Adds a remote offer to the model. Relations can be created later using "juju relate".
1168 :param: offer_url: Offer Url
1169 :param: model_name: Model name
1171 :raises ParseError if there's a problem parsing the offer_url
1172 :raises JujuError if remote offer includes and endpoint
1173 :raises JujuAPIError if the operation is not successful
1175 controller
= await self
.get_controller()
1176 model
= await controller
.get_model(model_name
)
1179 await model
.consume(offer_url
)
1181 await self
.disconnect_model(model
)
1182 await self
.disconnect_controller(controller
)
1184 async def destroy_model(self
, model_name
: str, total_timeout
: float = 1800):
1188 :param: model_name: Model name
1189 :param: total_timeout: Timeout
1192 controller
= await self
.get_controller()
1195 if not await self
.model_exists(model_name
, controller
=controller
):
1198 self
.log
.debug("Destroying model {}".format(model_name
))
1200 model
= await self
.get_model(controller
, model_name
)
1201 # Destroy machines that are manually provisioned
1202 # and still are in pending state
1203 await self
._destroy
_pending
_machines
(model
, only_manual
=True)
1204 await self
.disconnect_model(model
)
1206 await asyncio
.wait_for(
1207 self
._destroy
_model
(model_name
, controller
),
1208 timeout
=total_timeout
,
1210 except Exception as e
:
1211 if not await self
.model_exists(model_name
, controller
=controller
):
1216 await self
.disconnect_model(model
)
1217 await self
.disconnect_controller(controller
)
1219 async def _destroy_model(
1222 controller
: Controller
,
1225 Destroy model from controller
1227 :param: model: Model name to be removed
1228 :param: controller: Controller object
1229 :param: timeout: Timeout in seconds
1232 async def _destroy_model_gracefully(model_name
: str, controller
: Controller
):
1233 self
.log
.info(f
"Gracefully deleting model {model_name}")
1235 while model_name
in await controller
.list_models():
1237 await self
.resolve(model_name
)
1239 await controller
.destroy_model(model_name
, destroy_storage
=True)
1241 await asyncio
.sleep(5)
1242 self
.log
.info(f
"Model {model_name} deleted gracefully")
1244 async def _destroy_model_forcefully(model_name
: str, controller
: Controller
):
1245 self
.log
.info(f
"Forcefully deleting model {model_name}")
1246 while model_name
in await controller
.list_models():
1247 await controller
.destroy_model(
1248 model_name
, destroy_storage
=True, force
=True, max_wait
=60
1250 await asyncio
.sleep(5)
1251 self
.log
.info(f
"Model {model_name} deleted forcefully")
1255 await asyncio
.wait_for(
1256 _destroy_model_gracefully(model_name
, controller
), timeout
=120
1258 except asyncio
.TimeoutError
:
1259 await _destroy_model_forcefully(model_name
, controller
)
1260 except juju
.errors
.JujuError
as e
:
1261 if any("has been removed" in error
for error
in e
.errors
):
1263 if any("model not found" in error
for error
in e
.errors
):
1267 async def destroy_application(
1268 self
, model_name
: str, application_name
: str, total_timeout
: float
1273 :param: model_name: Model name
1274 :param: application_name: Application name
1275 :param: total_timeout: Timeout
1278 controller
= await self
.get_controller()
1282 model
= await self
.get_model(controller
, model_name
)
1284 "Destroying application {} in model {}".format(
1285 application_name
, model_name
1288 application
= self
._get
_application
(model
, application_name
)
1290 await application
.destroy()
1292 self
.log
.warning("Application not found: {}".format(application_name
))
1295 "Waiting for application {} to be destroyed in model {}...".format(
1296 application_name
, model_name
1299 if total_timeout
is None:
1300 total_timeout
= 3600
1301 end
= time
.time() + total_timeout
1302 while time
.time() < end
:
1303 if not self
._get
_application
(model
, application_name
):
1305 "The application {} was destroyed in model {} ".format(
1306 application_name
, model_name
1310 await asyncio
.sleep(5)
1312 "Timeout waiting for application {} to be destroyed in model {}".format(
1313 application_name
, model_name
1317 if model
is not None:
1318 await self
.disconnect_model(model
)
1319 await self
.disconnect_controller(controller
)
1321 async def _destroy_pending_machines(self
, model
: Model
, only_manual
: bool = False):
1323 Destroy pending machines in a given model
1325 :param: only_manual: Bool that indicates only manually provisioned
1326 machines should be destroyed (if True), or that
1327 all pending machines should be destroyed
1329 status
= await model
.get_status()
1330 for machine_id
in status
.machines
:
1331 machine_status
= status
.machines
[machine_id
]
1332 if machine_status
.agent_status
.status
== "pending":
1333 if only_manual
and not machine_status
.instance_id
.startswith("manual:"):
1335 machine
= model
.machines
[machine_id
]
1336 await machine
.destroy(force
=True)
1338 async def configure_application(
1339 self
, model_name
: str, application_name
: str, config
: dict = None
1341 """Configure application
1343 :param: model_name: Model name
1344 :param: application_name: Application name
1345 :param: config: Config to apply to the charm
1347 self
.log
.debug("Configuring application {}".format(application_name
))
1350 controller
= await self
.get_controller()
1353 model
= await self
.get_model(controller
, model_name
)
1354 application
= self
._get
_application
(
1356 application_name
=application_name
,
1358 await application
.set_config(config
)
1361 await self
.disconnect_model(model
)
1362 await self
.disconnect_controller(controller
)
1364 def handle_exception(self
, loop
, context
):
1365 # All unhandled exceptions by libjuju are handled here.
1368 async def health_check(self
, interval
: float = 300.0):
1370 Health check to make sure controller and controller_model connections are OK
1372 :param: interval: Time in seconds between checks
1377 controller
= await self
.get_controller()
1378 # self.log.debug("VCA is alive")
1379 except Exception as e
:
1380 self
.log
.error("Health check to VCA failed: {}".format(e
))
1382 await self
.disconnect_controller(controller
)
1383 await asyncio
.sleep(interval
)
1385 async def list_models(self
, contains
: str = None) -> [str]:
1386 """List models with certain names
1388 :param: contains: String that is contained in model name
1390 :retur: [models] Returns list of model names
1393 controller
= await self
.get_controller()
1395 models
= await controller
.list_models()
1397 models
= [model
for model
in models
if contains
in model
]
1400 await self
.disconnect_controller(controller
)
1402 async def list_offers(self
, model_name
: str) -> QueryApplicationOffersResults
:
1403 """List models with certain names
1405 :param: model_name: Model name
1407 :return: Returns list of offers
1410 controller
= await self
.get_controller()
1412 return await controller
.list_offers(model_name
)
1414 await self
.disconnect_controller(controller
)
1421 client_cert_data
: str,
1422 configuration
: Configuration
,
1424 credential_name
: str = None,
1427 Add a Kubernetes cloud to the controller
1429 Similar to the `juju add-k8s` command in the CLI
1431 :param: name: Name for the K8s cloud
1432 :param: configuration: Kubernetes configuration object
1433 :param: storage_class: Storage Class to use in the cloud
1434 :param: credential_name: Storage Class to use in the cloud
1437 if not storage_class
:
1438 raise Exception("storage_class must be a non-empty string")
1440 raise Exception("name must be a non-empty string")
1441 if not configuration
:
1442 raise Exception("configuration must be provided")
1444 endpoint
= configuration
.host
1445 credential
= self
.get_k8s_cloud_credential(
1450 credential
.attrs
[RBAC_LABEL_KEY_NAME
] = rbac_id
1451 cloud
= client
.Cloud(
1453 auth_types
=[credential
.auth_type
],
1455 ca_certificates
=[client_cert_data
],
1457 "operator-storage": storage_class
,
1458 "workload-storage": storage_class
,
1462 return await self
.add_cloud(
1463 name
, cloud
, credential
, credential_name
=credential_name
1466 def get_k8s_cloud_credential(
1468 configuration
: Configuration
,
1469 client_cert_data
: str,
1471 ) -> client
.CloudCredential
:
1473 # TODO: Test with AKS
1474 key
= None # open(configuration.key_file, "r").read()
1475 username
= configuration
.username
1476 password
= configuration
.password
1478 if client_cert_data
:
1479 attrs
["ClientCertificateData"] = client_cert_data
1481 attrs
["ClientKeyData"] = key
1483 if username
or password
:
1484 raise JujuInvalidK8sConfiguration("Cannot set both token and user/pass")
1485 attrs
["Token"] = token
1489 auth_type
= "oauth2"
1490 if client_cert_data
:
1491 auth_type
= "oauth2withcert"
1493 raise JujuInvalidK8sConfiguration(
1494 "missing token for auth type {}".format(auth_type
)
1499 "credential for user {} has empty password".format(username
)
1501 attrs
["username"] = username
1502 attrs
["password"] = password
1503 if client_cert_data
:
1504 auth_type
= "userpasswithcert"
1506 auth_type
= "userpass"
1507 elif client_cert_data
and token
:
1508 auth_type
= "certificate"
1510 raise JujuInvalidK8sConfiguration("authentication method not supported")
1511 return client
.CloudCredential(auth_type
=auth_type
, attrs
=attrs
)
1513 async def add_cloud(
1517 credential
: CloudCredential
= None,
1518 credential_name
: str = None,
1521 Add cloud to the controller
1523 :param: name: Name of the cloud to be added
1524 :param: cloud: Cloud object
1525 :param: credential: CloudCredentials object for the cloud
1526 :param: credential_name: Credential name.
1527 If not defined, cloud of the name will be used.
1529 controller
= await self
.get_controller()
1531 _
= await controller
.add_cloud(name
, cloud
)
1533 await controller
.add_credential(
1534 credential_name
or name
, credential
=credential
, cloud
=name
1536 # Need to return the object returned by the controller.add_cloud() function
1537 # I'm returning the original value now until this bug is fixed:
1538 # https://github.com/juju/python-libjuju/issues/443
1541 await self
.disconnect_controller(controller
)
1543 async def remove_cloud(self
, name
: str):
1547 :param: name: Name of the cloud to be removed
1549 controller
= await self
.get_controller()
1551 await controller
.remove_cloud(name
)
1552 except juju
.errors
.JujuError
as e
:
1553 if len(e
.errors
) == 1 and f
'cloud "{name}" not found' == e
.errors
[0]:
1554 self
.log
.warning(f
"Cloud {name} not found, so it could not be deleted.")
1558 await self
.disconnect_controller(controller
)
1560 @retry(attempts
=20, delay
=5, fallback
=JujuLeaderUnitNotFound())
1561 async def _get_leader_unit(self
, application
: Application
) -> Unit
:
1563 for u
in application
.units
:
1564 if await u
.is_leader_from_status():
1571 async def get_cloud_credentials(self
, cloud
: Cloud
) -> typing
.List
:
1573 Get cloud credentials
1575 :param: cloud: Cloud object. The returned credentials will be from this cloud.
1577 :return: List of credentials object associated to the specified cloud
1580 controller
= await self
.get_controller()
1582 facade
= client
.CloudFacade
.from_connection(controller
.connection())
1583 cloud_cred_tag
= tag
.credential(
1584 cloud
.name
, self
.vca_connection
.data
.user
, cloud
.credential_name
1586 params
= [client
.Entity(cloud_cred_tag
)]
1587 return (await facade
.Credential(params
)).results
1589 await self
.disconnect_controller(controller
)
1591 async def check_application_exists(self
, model_name
, application_name
) -> bool:
1592 """Check application exists
1594 :param: model_name: Model Name
1595 :param: application_name: Application Name
1601 controller
= await self
.get_controller()
1603 model
= await self
.get_model(controller
, model_name
)
1605 "Checking if application {} exists in model {}".format(
1606 application_name
, model_name
1609 return self
._get
_application
(model
, application_name
) is not None
1612 await self
.disconnect_model(model
)
1613 await self
.disconnect_controller(controller
)