cluster_uuid,
kdu_instance,
filter=db_dict["filter"],
- vca_id=kwargs.get("vca_id")
+ vca_id=kwargs.get("vca_id"),
)
return True
async def scale(
- self,
- kdu_instance: str,
- scale: int,
- resource_name: str,
- total_timeout: float = 1800,
- **kwargs,
+ self,
+ kdu_instance: str,
+ scale: int,
+ resource_name: str,
+ total_timeout: float = 1800,
+ **kwargs,
) -> bool:
"""Scale an application in a model
model_name=kdu_instance,
application_name=resource_name,
scale=scale,
- total_timeout=total_timeout
+ total_timeout=total_timeout,
)
except Exception as e:
error_msg = "Error scaling application {} in kdu instance {}: {}".format(
- resource_name, kdu_instance, e)
+ resource_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,
- **kwargs,
+ self,
+ resource_name: str,
+ kdu_instance: str,
+ **kwargs,
) -> int:
"""Get an application scale count
return len(status.applications[resource_name].units)
except Exception as e:
error_msg = "Error getting scale count from application {} in kdu instance {}: {}".format(
- resource_name, kdu_instance, e)
+ resource_name, kdu_instance, e
+ )
self.log.error(error_msg)
raise K8sException(message=error_msg)
"status is not completed: {} output: {}".format(status, output)
)
if self.on_update_db:
- await self.on_update_db(cluster_uuid, kdu_instance, filter=db_dict["filter"])
+ await self.on_update_db(
+ cluster_uuid, kdu_instance, filter=db_dict["filter"]
+ )
return output
try:
for model_name in vcastatus:
# Adding executed actions
- vcastatus[model_name]["executedActions"] = \
- await libjuju.get_executed_actions(kdu_instance)
+ vcastatus[model_name][
+ "executedActions"
+ ] = await libjuju.get_executed_actions(kdu_instance)
for application in vcastatus[model_name]["applications"]:
# Adding application actions
- vcastatus[model_name]["applications"][application]["actions"] = \
- await libjuju.get_actions(application, kdu_instance)
+ vcastatus[model_name]["applications"][application][
+ "actions"
+ ] = await libjuju.get_actions(application, kdu_instance)
# Adding application configs
- vcastatus[model_name]["applications"][application]["configs"] = \
- await libjuju.get_application_configs(kdu_instance, application)
+ vcastatus[model_name]["applications"][application][
+ "configs"
+ ] = await libjuju.get_application_configs(kdu_instance, application)
except Exception as e:
self.log.debug("Error in updating vca status: {}".format(str(e)))