+ async def scale(
+ self,
+ kdu_instance: str,
+ scale: int,
+ resource_name: str,
+ total_timeout: float = 1800,
+ namespace: str = None,
+ **kwargs,
+ ) -> bool:
+ """Scale an application in a model
+
+ :param: kdu_instance str: KDU instance name
+ :param: scale int: Scale to which to set the application
+ :param: resource_name str: The application name in the Juju Bundle
+ :param: timeout float: The time, in seconds, to wait for the install
+ to finish
+ :param namespace str: The namespace (model) where the Bundle was deployed
+ :param kwargs: Additional parameters
+ vca_id (str): VCA ID
+
+ :return: If successful, returns True
+ """
+
+ model_name = self._obtain_namespace(
+ kdu_instance=kdu_instance, namespace=namespace
+ )
+ try:
+ libjuju = await self._get_libjuju(kwargs.get("vca_id"))
+ await libjuju.scale_application(
+ model_name=model_name,
+ application_name=resource_name,
+ scale=scale,
+ total_timeout=total_timeout,
+ )
+ except Exception as e:
+ error_msg = "Error scaling application {} of the model {} of the kdu instance {}: {}".format(
+ resource_name, model_name, kdu_instance, e
+ )
+ self.log.error(error_msg)
+ raise K8sException(message=error_msg)
+ return True
+
+ async def get_scale_count(
+ self,
+ resource_name: str,
+ kdu_instance: str,
+ namespace: str = None,
+ **kwargs,
+ ) -> int:
+ """Get an application scale count
+
+ :param: resource_name str: The application name in the Juju Bundle
+ :param: kdu_instance str: KDU instance name
+ :param namespace str: The namespace (model) where the Bundle was deployed
+ :param kwargs: Additional parameters
+ vca_id (str): VCA ID
+ :return: Return application instance count
+ """
+
+ model_name = self._obtain_namespace(
+ kdu_instance=kdu_instance, namespace=namespace
+ )
+ try:
+ libjuju = await self._get_libjuju(kwargs.get("vca_id"))
+ status = await libjuju.get_model_status(model_name=model_name)
+ return len(status.applications[resource_name].units)
+ except Exception as e:
+ error_msg = (
+ f"Error getting scale count from application {resource_name} of the model {model_name} of "
+ f"the kdu instance {kdu_instance}: {e}"
+ )
+ self.log.error(error_msg)
+ raise K8sException(message=error_msg)
+
+ async def instances_list(self, cluster_uuid: str) -> list: