- # Private methods
- async def add_k8s(self, cloud_name: str, credentials: str,) -> bool:
- """Add a k8s cloud to Juju
-
- Adds a Kubernetes cloud to Juju, so it can be bootstrapped with a
- Juju Controller.
-
- :param cloud_name str: The name of the cloud to add.
- :param credentials dict: A dictionary representing the output of
- `kubectl config view --raw`.
-
- :returns: True if successful, otherwise raises an exception.
- """
-
- cmd = [self.juju_command, "add-k8s", "--local", cloud_name]
- self.log.debug(cmd)
-
- process = await asyncio.create_subprocess_exec(
- *cmd,
- stdout=asyncio.subprocess.PIPE,
- stderr=asyncio.subprocess.PIPE,
- stdin=asyncio.subprocess.PIPE,
- )
-
- # Feed the process the credentials
- process.stdin.write(credentials.encode("utf-8"))
- await process.stdin.drain()
- process.stdin.close()
-
- _stdout, stderr = await process.communicate()
-
- return_code = process.returncode
-
- self.log.debug("add-k8s return code: {}".format(return_code))
-
- if return_code > 0:
- raise Exception(stderr)
-
- return True
-
- async def add_model(
- self, model_name: str, cluster_uuid: str, controller: Controller
- ) -> Model:
- """Adds a model to the controller
-
- Adds a new model to the Juju controller
-
- :param model_name str: The name of the model to add.
- :param cluster_uuid str: ID of the cluster.
- :param controller: Controller object in which the model will be added
- :returns: The juju.model.Model object of the new model upon success or
- raises an exception.
- """
-
- self.log.debug(
- "Adding model '{}' to cluster_uuid '{}'".format(model_name, cluster_uuid)
- )
- model = None
- try:
- if self.juju_public_key is not None:
- model = await controller.add_model(
- model_name, config={"authorized-keys": self.juju_public_key}
- )
- else:
- model = await controller.add_model(model_name)
- except Exception as ex:
- self.log.debug(ex)
- self.log.debug("Caught exception: {}".format(ex))
- pass
-
- return model
-
- async def bootstrap(
- self, cloud_name: str, cluster_uuid: str, loadbalancer: bool
- ) -> bool:
- """Bootstrap a Kubernetes controller
-
- Bootstrap a Juju controller inside the Kubernetes cluster
-
- :param cloud_name str: The name of the cloud.
- :param cluster_uuid str: The UUID of the cluster to bootstrap.
- :param loadbalancer bool: If the controller should use loadbalancer or not.
- :returns: True upon success or raises an exception.
- """
-
- if not loadbalancer:
- cmd = [self.juju_command, "bootstrap", cloud_name, cluster_uuid]
- else:
- """
- For public clusters, specify that the controller service is using a
- LoadBalancer.
- """
- cmd = [
- self.juju_command,
- "bootstrap",
- cloud_name,
- cluster_uuid,
- "--config",
- "controller-service-type=loadbalancer",
- ]
-
- self.log.debug(
- "Bootstrapping controller {} in cloud {}".format(cluster_uuid, cloud_name)
- )
-
- process = await asyncio.create_subprocess_exec(
- *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
- )
-
- _stdout, stderr = await process.communicate()
-
- return_code = process.returncode
-
- if return_code > 0:
- #
- if b"already exists" not in stderr:
- raise Exception(stderr)
-
- return True
-
- async def destroy_controller(self, cluster_uuid: str) -> bool:
- """Destroy a Kubernetes controller
-
- Destroy an existing Kubernetes controller.
-
- :param cluster_uuid str: The UUID of the cluster to bootstrap.
- :returns: True upon success or raises an exception.
- """
- cmd = [
- self.juju_command,
- "destroy-controller",
- "--destroy-all-models",
- "--destroy-storage",
- "-y",
- cluster_uuid,
- ]
-
- process = await asyncio.create_subprocess_exec(
- *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
- )
-
- _stdout, stderr = await process.communicate()
-
- return_code = process.returncode
-
- if return_code > 0:
- #
- if "already exists" not in stderr:
- raise Exception(stderr)
-