from juju.controller import Controller
from juju.model import Model
from juju.errors import JujuAPIError, JujuError
-
-import logging
+from n2vc.exceptions import K8sException
from n2vc.k8s_conn import K8sConnector
db: object,
kubectl_command: str = '/usr/bin/kubectl',
juju_command: str = '/usr/bin/juju',
- log=None,
+ log: object = None,
on_update_db=None,
):
"""
)
self.fs = fs
- self.info('Initializing K8S Juju connector')
+ self.log.debug('Initializing K8S Juju connector')
self.authenticated = False
self.models = {}
- self.log = logging.getLogger(__name__)
self.juju_command = juju_command
self.juju_secret = ""
- self.info('K8S Juju connector initialized')
+ self.log.debug('K8S Juju connector initialized')
"""Initialization"""
async def init_env(
# Name the new k8s cloud
k8s_cloud = "k8s-{}".format(cluster_uuid)
- print("Adding k8s cloud {}".format(k8s_cloud))
+ self.log.debug("Adding k8s cloud {}".format(k8s_cloud))
await self.add_k8s(k8s_cloud, k8s_creds)
# Bootstrap Juju controller
- print("Bootstrapping...")
+ self.log.debug("Bootstrapping...")
await self.bootstrap(k8s_cloud, cluster_uuid, loadbalancer)
- print("Bootstrap done.")
+ self.log.debug("Bootstrap done.")
# Get the controller information
# Parse ~/.local/share/juju/controllers.yaml
# controllers.testing.api-endpoints|ca-cert|uuid
- print("Getting controller endpoints")
+ self.log.debug("Getting controller endpoints")
with open(os.path.expanduser(
"~/.local/share/juju/controllers.yaml"
)) as f:
# Parse ~/.local/share/juju/accounts
# controllers.testing.user|password
- print("Getting accounts")
+ self.log.debug("Getting accounts")
with open(os.path.expanduser(
"~/.local/share/juju/accounts.yaml"
)) as f:
self.juju_user = controller['user']
self.juju_secret = controller['password']
- print("user: {}".format(self.juju_user))
- print("secret: {}".format(self.juju_secret))
- print("endpoint: {}".format(self.juju_endpoint))
- print("ca-cert: {}".format(self.juju_ca_cert))
-
# raise Exception("EOL")
self.juju_public_key = None
# Store the cluster configuration so it
# can be used for subsequent calls
- print("Setting config")
+ self.log.debug("Setting config")
await self.set_config(cluster_uuid, config)
else:
# Destroy the model
namespace = self.get_namespace(cluster_uuid)
if await self.has_model(namespace):
- print("[reset] Destroying model")
+ self.log.debug("[reset] Destroying model")
await self.controller.destroy_model(
namespace,
destroy_storage=True
)
# Disconnect from the controller
- print("[reset] Disconnecting controller")
+ self.log.debug("[reset] Disconnecting controller")
await self.logout()
# Destroy the controller (via CLI)
- print("[reset] Destroying controller")
+ self.log.debug("[reset] Destroying controller")
await self.destroy_controller(cluster_uuid)
- print("[reset] Removing k8s cloud")
+ self.log.debug("[reset] Removing k8s cloud")
k8s_cloud = "k8s-{}".format(cluster_uuid)
await self.remove_cloud(k8s_cloud)
except Exception as ex:
- print("Caught exception during reset: {}".format(ex))
+ self.log.debug("Caught exception during reset: {}".format(ex))
return True
timeout: float = 300,
params: dict = None,
db_dict: dict = None,
- kdu_name: str = None
+ kdu_name: str = None,
+ namespace: str = None
) -> bool:
"""Install a bundle
to finish
:param params dict: Key-value pairs of instantiation parameters
:param kdu_name: Name of the KDU instance to be installed
+ :param namespace: K8s namespace to use for the KDU instance
:return: If successful, returns ?
"""
if not self.authenticated:
- print("[install] Logging in to the controller")
+ self.log.debug("[install] Logging in to the controller")
await self.login(cluster_uuid)
##
- <URL_where_to_fetch_juju_bundle>
"""
+ previous_workdir = os.getcwd()
+
bundle = kdu_model
if kdu_model.startswith("cs:"):
bundle = kdu_model
# Download the file
pass
else:
- # Local file
+ new_workdir = kdu_model.strip(kdu_model.split("/")[-1])
- # if kdu_model.endswith(".tar.gz") or kdu_model.endswith(".tgz")
- # Uncompress temporarily
- # bundle = <uncompressed file>
- pass
+ os.chdir(new_workdir)
+
+ bundle = "local:{}".format(kdu_model)
if not bundle:
# Raise named exception that the bundle could not be found
raise Exception()
- print("[install] deploying {}".format(bundle))
+ self.log.debug("[install] deploying {}".format(bundle))
await model.deploy(bundle)
# Get the application
if atomic:
# applications = model.applications
- print("[install] Applications: {}".format(model.applications))
+ self.log.debug("[install] Applications: {}".format(model.applications))
for name in model.applications:
- print("[install] Waiting for {} to settle".format(name))
+ self.log.debug("[install] Waiting for {} to settle".format(name))
application = model.applications[name]
try:
# It's not enough to wait for all units to be active;
# the application status needs to be active as well.
- print("Waiting for all units to be active...")
+ self.log.debug("Waiting for all units to be active...")
await model.block_until(
lambda: all(
unit.agent_status == 'idle'
),
timeout=timeout
)
- print("All units active.")
+ self.log.debug("All units active.")
- except concurrent.futures._base.TimeoutError:
- print("[install] Timeout exceeded; resetting cluster")
+ except concurrent.futures._base.TimeoutError: # TODO use asyncio.TimeoutError
+ os.chdir(previous_workdir)
+ self.log.debug("[install] Timeout exceeded; resetting cluster")
await self.reset(cluster_uuid)
return False
# Wait for the application to be active
if model.is_connected():
- print("[install] Disconnecting model")
+ self.log.debug("[install] Disconnecting model")
await model.disconnect()
+ os.chdir(previous_workdir)
+
return kdu_instance
raise Exception("Unable to install")
"""
# TODO: This should be returned in an agreed-upon format
for name in bundle['applications']:
- print(model.applications)
+ self.log.debug(model.applications)
application = model.applications[name]
- print(application)
+ self.log.debug(application)
path = bundle['applications'][name]['charm']
return True
+ async def exec_primitive(
+ self,
+ cluster_uuid: str = None,
+ kdu_instance: str = None,
+ primitive_name: str = None,
+ timeout: float = 300,
+ params: dict = None,
+ db_dict: dict = None,
+ ) -> str:
+ """Exec primitive (Juju action)
+
+ :param cluster_uuid str: The UUID of the cluster
+ :param kdu_instance str: The unique name of the KDU instance
+ :param primitive_name: Name of action that will be executed
+ :param timeout: Timeout for action execution
+ :param params: Dictionary of all the parameters needed for the action
+ :db_dict: Dictionary for any additional data
+
+ :return: Returns the output of the action
+ """
+ if not self.authenticated:
+ self.log.debug("[exec_primitive] Connecting to controller")
+ await self.login(cluster_uuid)
+
+ if not params or "application-name" not in params:
+ raise K8sException("Missing application-name argument, \
+ argument needed for K8s actions")
+ try:
+ self.log.debug("[exec_primitive] Getting model "
+ "kdu_instance: {}".format(kdu_instance))
+
+ model = await self.get_model(kdu_instance, cluster_uuid)
+
+ application_name = params["application-name"]
+ application = model.applications[application_name]
+
+ actions = await application.get_actions()
+ if primitive_name not in actions:
+ raise K8sException("Primitive {} not found".format(primitive_name))
+
+ unit = None
+ for u in application.units:
+ if await u.is_leader_from_status():
+ unit = u
+ break
+
+ if unit is None:
+ raise K8sException("No leader unit found to execute action")
+
+ self.log.debug("[exec_primitive] Running action: {}".format(primitive_name))
+ action = await unit.run_action(primitive_name, **params)
+
+ output = await model.get_action_output(action_uuid=action.entity_id)
+ status = await model.get_action_status(uuid_or_prefix=action.entity_id)
+
+ status = (
+ status[action.entity_id] if action.entity_id in status else "failed"
+ )
+
+ if status != "completed":
+ raise K8sException("status is not completed: {} output: {}".format(status, output))
+
+ return output
+
+ except Exception as e:
+ error_msg = "Error executing primitive {}: {}".format(primitive_name, e)
+ self.log.error(error_msg)
+ raise K8sException(message=error_msg)
+
"""Introspection"""
async def inspect_kdu(
self,
"""
cmd = [self.juju_command, "add-k8s", "--local", cloud_name]
- print(cmd)
+ self.log.debug(cmd)
process = await asyncio.create_subprocess_exec(
*cmd,
return_code = process.returncode
- print("add-k8s return code: {}".format(return_code))
+ self.log.debug("add-k8s return code: {}".format(return_code))
if return_code > 0:
raise Exception(stderr)
"""
cmd = [self.juju_command, "bootstrap", cloud_name, cluster_uuid, "--config", "controller-service-type=loadbalancer"]
- print("Bootstrapping controller {} in cloud {}".format(
+ self.log.debug("Bootstrapping controller {} in cloud {}".format(
cluster_uuid, cloud_name
))
model = None
models = await self.controller.list_models()
- self.log.debug(models)
if model_name in models:
self.log.debug("Found model: {}".format(model_name))
model = await self.controller.get_model(
self.authenticated = True
self.log.debug("JujuApi: Logged into controller")
except Exception as ex:
- print(ex)
+ self.log.debug(ex)
self.log.debug("Caught exception: {}".format(ex))
pass
else:
async def logout(self):
"""Logout of the Juju controller."""
- print("[logout]")
+ self.log.debug("[logout]")
if not self.authenticated:
return False
for model in self.models:
- print("Logging out of model {}".format(model))
+ self.log.debug("Logging out of model {}".format(model))
await self.models[model].disconnect()
if self.controller:
cluster_config = "{}/{}.yaml".format(self.fs.path, cluster_uuid)
if not os.path.exists(cluster_config):
- print("Writing config to {}".format(cluster_config))
+ self.log.debug("Writing config to {}".format(cluster_config))
with open(cluster_config, 'w') as f:
f.write(yaml.dump(config, Dumper=yaml.Dumper))
projects / osm / N2VC.git / blobdiff