# See the License for the specific language governing permissions and
# limitations under the License.
+import asyncio
import concurrent
from .exceptions import NotImplemented
+import io
import juju
# from juju.bundle import BundleHandler
from juju.controller import Controller
from juju.model import Model
from juju.errors import JujuAPIError, JujuError
-import logging
-
from n2vc.k8s_conn import K8sConnector
import os
# import re
# import ssl
-import subprocess
# from .vnf import N2VC
import uuid
class K8sJujuConnector(K8sConnector):
+
def __init__(
self,
- fs,
- kubectl_command='/usr/bin/kubectl',
- log=None
+ fs: object,
+ db: object,
+ kubectl_command: str = '/usr/bin/kubectl',
+ juju_command: str = '/usr/bin/juju',
+ log: object = None,
+ on_update_db=None,
):
"""
# parent class
K8sConnector.__init__(
self,
- kubectl_command=kubectl_command,
- fs=fs,
+ db,
log=log,
+ on_update_db=on_update_db,
)
- self.info('Initializing K8S Juju connector')
+ self.fs = fs
+ self.log.debug('Initializing K8S Juju connector')
self.authenticated = False
self.models = {}
- self.log = logging.getLogger(__name__)
- self.info('K8S Juju connector initialized')
+
+ self.juju_command = juju_command
+ self.juju_secret = ""
+
+ self.log.debug('K8S Juju connector initialized')
"""Initialization"""
async def init_env(
self,
- k8s_creds: dict,
+ k8s_creds: str,
namespace: str = 'kube-system',
reuse_cluster_uuid: str = None,
- ) -> str:
- """Initialize a Kubernetes environment
-
- :param k8s_creds dict: A dictionary containing the Kubernetes cluster
- configuration
- :param namespace str: The Kubernetes namespace to initialize
+ ) -> (str, bool):
+ """
+ It prepares a given K8s cluster environment to run Juju bundles.
- :return: UUID of the k8s context or raises an exception
+ :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid '.kube/config'
+ :param namespace: optional namespace to be used for juju. By default, 'kube-system' will be used
+ :param reuse_cluster_uuid: existing cluster uuid for reuse
+ :return: uuid of the K8s cluster and True if connector has installed some software in the cluster
+ (on error, an exception will be raised)
"""
"""Bootstrapping
Bootstrapping cannot be done, by design, through the API. We need to
use the CLI tools.
"""
- # TODO: The path may change
- jujudir = "/snap/bin"
-
- self.k8scli = "{}/juju".format(jujudir)
"""
WIP: Workflow
# TODO: Pull info from db based on the namespace #
##################################################
+ ###################################################
+ # TODO: Make it idempotent, calling add-k8s and #
+ # bootstrap whenever reuse_cluster_uuid is passed #
+ # as parameter #
+ # `init_env` is called to initialize the K8s #
+ # cluster for juju. If this initialization fails, #
+ # it can be called again by LCM with the param #
+ # reuse_cluster_uuid, e.g. to try to fix it. #
+ ###################################################
+
if not reuse_cluster_uuid:
# This is a new cluster, so bootstrap it
cluster_uuid = str(uuid.uuid4())
- # Add k8s cloud to Juju (unless it's microk8s)
+ # Is a local k8s cluster?
+ localk8s = self.is_local_k8s(k8s_creds)
- # Does the kubeconfig contain microk8s?
- microk8s = self.is_microk8s_by_credentials(k8s_creds)
+ # If the k8s is external, the juju controller needs a loadbalancer
+ loadbalancer = False if localk8s else True
- if not microk8s:
- # Name the new k8s cloud
- k8s_cloud = "{}-k8s".format(namespace)
-
- await self.add_k8s(k8s_cloud, k8s_creds)
-
- # Bootstrap Juju controller
- self.bootstrap(k8s_cloud, cluster_uuid)
- else:
- # k8s_cloud = 'microk8s-test'
- k8s_cloud = "{}-k8s".format(namespace)
+ # Name the new k8s cloud
+ k8s_cloud = "k8s-{}".format(cluster_uuid)
- await self.add_k8s(k8s_cloud, k8s_creds)
+ self.log.debug("Adding k8s cloud {}".format(k8s_cloud))
+ await self.add_k8s(k8s_cloud, k8s_creds)
- await self.bootstrap(k8s_cloud, cluster_uuid)
+ # Bootstrap Juju controller
+ self.log.debug("Bootstrapping...")
+ await self.bootstrap(k8s_cloud, cluster_uuid, loadbalancer)
+ self.log.debug("Bootstrap done.")
# Get the controller information
# Parse ~/.local/share/juju/controllers.yaml
# controllers.testing.api-endpoints|ca-cert|uuid
+ 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
+ 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
'secret': self.juju_secret,
'cacert': self.juju_ca_cert,
'namespace': namespace,
- 'microk8s': microk8s,
+ 'loadbalancer': loadbalancer,
}
# Store the cluster configuration so it
# can be used for subsequent calls
+ self.log.debug("Setting config")
await self.set_config(cluster_uuid, config)
else:
# Login to the k8s cluster
if not self.authenticated:
- await self.login()
+ await self.login(cluster_uuid)
# We're creating a new cluster
- print("Getting model {}".format(self.get_namespace(cluster_uuid)))
- model = await self.get_model(self.get_namespace(cluster_uuid))
+ #print("Getting model {}".format(self.get_namespace(cluster_uuid), cluster_uuid=cluster_uuid))
+ #model = await self.get_model(
+ # self.get_namespace(cluster_uuid),
+ # cluster_uuid=cluster_uuid
+ #)
- # Disconnect from the model
- if model and model.is_connected():
- await model.disconnect()
+ ## Disconnect from the model
+ #if model and model.is_connected():
+ # await model.disconnect()
- return cluster_uuid
+ return cluster_uuid, True
"""Repo Management"""
async def repo_add(
):
raise NotImplemented()
- """Reset"""
- async def reset(
+ async def synchronize_repos(
self,
cluster_uuid: str,
+ name: str
+ ):
+ """
+ Returns None as currently add_repo is not implemented
+ """
+ return None
+
+ """Reset"""
+ async def reset(
+ self,
+ cluster_uuid: str,
+ force: bool = False,
+ uninstall_sw: bool = False
) -> bool:
"""Reset a cluster
try:
if not self.authenticated:
- await self.login()
+ await self.login(cluster_uuid)
if self.controller.is_connected():
# 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")
- await self.controller.disconnect()
+ 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)
- """Remove the k8s cloud
-
- Only remove the k8s cloud if it's not a microk8s cloud,
- since microk8s is a built-in cloud type.
- """
- # microk8s = self.is_microk8s_by_cluster_uuid(cluster_uuid)
- # if not microk8s:
- print("[reset] Removing k8s cloud")
- namespace = self.get_namespace(cluster_uuid)
- k8s_cloud = "{}-k8s".format(namespace)
+ 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
"""Deployment"""
+
async def install(
self,
cluster_uuid: str,
kdu_model: str,
atomic: bool = True,
- timeout: int = None,
+ timeout: float = 300,
params: dict = None,
- ) -> str:
+ db_dict: dict = None,
+ kdu_name: str = None
+ ) -> bool:
"""Install a bundle
:param cluster_uuid str: The UUID of the cluster to install to
:param timeout int: The time, in seconds, to wait for the install
to finish
:param params dict: Key-value pairs of instantiation parameters
+ :param kdu_name: Name of the KDU instance to be installed
:return: If successful, returns ?
"""
if not self.authenticated:
- print("[install] Logging in to the controller")
- await self.login()
+ self.log.debug("[install] Logging in to the controller")
+ await self.login(cluster_uuid)
##
- # Get or create the model, based on the namespace the cluster was
- # instantiated with.
- namespace = self.get_namespace(cluster_uuid)
- model = await self.get_model(namespace)
- if not model:
- # Create the new model
- model = await self.add_model(namespace)
+ # Get or create the model, based on the NS
+ # uuid.
+ if kdu_name:
+ kdu_instance = "{}-{}".format(kdu_name, db_dict["filter"]["_id"])
+ else:
+ kdu_instance = db_dict["filter"]["_id"]
+
+ self.log.debug("Checking for model named {}".format(kdu_instance))
+
+ # Create the new model
+ self.log.debug("Adding model: {}".format(kdu_instance))
+ model = await self.add_model(kdu_instance, cluster_uuid=cluster_uuid)
if model:
# TODO: Instantiation parameters
- print("[install] deploying {}".format(kdu_model))
- await model.deploy(kdu_model)
+ """
+ "Juju bundle that models the KDU, in any of the following ways:
+ - <juju-repo>/<juju-bundle>
+ - <juju-bundle folder under k8s_models folder in the package>
+ - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder in the package>
+ - <URL_where_to_fetch_juju_bundle>
+ """
+
+ previous_workdir = os.getcwd()
+
+ bundle = kdu_model
+ if kdu_model.startswith("cs:"):
+ bundle = kdu_model
+ elif kdu_model.startswith("http"):
+ # Download the file
+ pass
+ else:
+ new_workdir = kdu_model.strip(kdu_model.split("/")[-1])
+
+ os.chdir(new_workdir)
+
+ bundle = "local:{}".format(kdu_model)
+
+ if not bundle:
+ # Raise named exception that the bundle could not be found
+ raise Exception()
+
+ 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")
+ 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()
- return True
+ os.chdir(previous_workdir)
+
+ return kdu_instance
raise Exception("Unable to install")
async def instances_list(
initial release.
"""
namespace = self.get_namespace(cluster_uuid)
- model = await self.get_model(namespace)
+ model = await self.get_model(namespace, cluster_uuid=cluster_uuid)
with open(kdu_model, 'r') as f:
- bundle = yaml.load(f, Loader=yaml.FullLoader)
+ bundle = yaml.safe_load(f)
"""
{
"""
# 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']
async def uninstall(
self,
cluster_uuid: str,
- kdu_instance: str,
+ kdu_instance: str
) -> bool:
"""Uninstall a KDU instance
- :param cluster_uuid str: The UUID of the cluster to uninstall
+ :param cluster_uuid str: The UUID of the cluster
:param kdu_instance str: The unique name of the KDU instance
:return: Returns True if successful, or raises an exception
"""
- removed = False
-
- # Remove an application from the model
- model = await self.get_model(self.get_namespace(cluster_uuid))
-
- if model:
- # Get the application
- if kdu_instance not in model.applications:
- # TODO: Raise a named exception
- raise Exception("Application not found.")
+ if not self.authenticated:
+ self.log.debug("[uninstall] Connecting to controller")
+ await self.login(cluster_uuid)
- application = model.applications[kdu_instance]
+ self.log.debug("[uninstall] Destroying model")
- # Destroy the application
- await application.destroy()
+ await self.controller.destroy_models(kdu_instance)
- # TODO: Verify removal
+ self.log.debug("[uninstall] Model destroyed and disconnecting")
+ await self.logout()
- removed = True
- return removed
+ return True
"""Introspection"""
async def inspect_kdu(
kdu = {}
with open(kdu_model, 'r') as f:
- bundle = yaml.load(f, Loader=yaml.FullLoader)
+ bundle = yaml.safe_load(f)
"""
{
"""
status = {}
- model = await self.get_model(self.get_namespace(cluster_uuid))
+ model = await self.get_model(self.get_namespace(cluster_uuid), cluster_uuid=cluster_uuid)
# model = await self.get_model_by_uuid(cluster_uuid)
if model:
async def add_k8s(
self,
cloud_name: str,
- credentials: dict,
+ credentials: str,
) -> bool:
"""Add a k8s cloud to Juju
:returns: True if successful, otherwise raises an exception.
"""
- cmd = [self.k8scli, "add-k8s", "--local", cloud_name]
- p = subprocess.run(
- cmd,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- input=yaml.dump(credentials, Dumper=yaml.Dumper),
- encoding='ascii'
+ 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,
)
- retcode = p.returncode
- if retcode > 0:
- raise Exception(p.stderr)
+ # 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
+ model_name: str,
+ cluster_uuid: str,
) -> juju.model.Model:
"""Adds a model to the controller
raises an exception.
"""
if not self.authenticated:
- await self.login()
+ await self.login(cluster_uuid)
+
+ self.log.debug("Adding model '{}' to cluster_uuid '{}'".format(model_name, cluster_uuid))
+ try:
+ model = await self.controller.add_model(
+ model_name,
+ config={'authorized-keys': self.juju_public_key}
+ )
+ except Exception as ex:
+ self.log.debug(ex)
+ self.log.debug("Caught exception: {}".format(ex))
+ pass
- model = await self.controller.add_model(
- model_name,
- config={'authorized-keys': self.juju_public_key}
- )
return model
async def bootstrap(
self,
cloud_name: str,
- cluster_uuid: str
+ cluster_uuid: str,
+ loadbalancer: bool
) -> bool:
"""Bootstrap a Kubernetes controller
: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.
"""
- cmd = [self.k8scli, "bootstrap", cloud_name, cluster_uuid]
- print("Bootstrapping controller {} in cloud {}".format(
+
+ 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
))
- p = subprocess.run(
- cmd,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- encoding='ascii'
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
)
- retcode = p.returncode
- if retcode > 0:
+ stdout, stderr = await process.communicate()
+
+ return_code = process.returncode
+
+ if return_code > 0:
#
- if 'already exists' not in p.stderr:
- raise Exception(p.stderr)
+ if b'already exists' not in stderr:
+ raise Exception(stderr)
return True
:returns: True upon success or raises an exception.
"""
cmd = [
- self.k8scli,
+ self.juju_command,
"destroy-controller",
"--destroy-all-models",
"--destroy-storage",
cluster_uuid
]
- p = subprocess.run(
- cmd,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- encoding='ascii'
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
)
- retcode = p.returncode
- if retcode > 0:
+ stdout, stderr = await process.communicate()
+
+ return_code = process.returncode
+
+ if return_code > 0:
#
- if 'already exists' not in p.stderr:
- raise Exception(p.stderr)
+ if 'already exists' not in stderr:
+ raise Exception(stderr)
def get_config(
self,
cluster_config = "{}/{}.yaml".format(self.fs.path, cluster_uuid)
if os.path.exists(cluster_config):
with open(cluster_config, 'r') as f:
- config = yaml.load(f.read(), Loader=yaml.FullLoader)
+ config = yaml.safe_load(f.read())
return config
else:
raise Exception(
async def get_model(
self,
model_name: str,
+ cluster_uuid: str,
) -> juju.model.Model:
"""Get a model from the Juju Controller.
:return The juju.model.Model object if found, or None.
"""
if not self.authenticated:
- await self.login()
+ await self.login(cluster_uuid)
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(
model_name
)
return True
return False
- def is_microk8s_by_cluster_uuid(
- self,
- cluster_uuid: str,
- ) -> bool:
- """Check if a cluster is micro8s
-
- Checks if a cluster is running microk8s
-
- :param cluster_uuid str: The UUID of the cluster
- :returns: A boolean if the cluster is running microk8s
- """
- config = self.get_config(cluster_uuid)
- return config['microk8s']
-
- def is_microk8s_by_credentials(
+ def is_local_k8s(
self,
- credentials: dict,
+ credentials: str,
) -> bool:
- """Check if a cluster is micro8s
+ """Check if a cluster is local
- Checks if a cluster is running microk8s
+ Checks if a cluster is running in the local host
:param credentials dict: A dictionary containing the k8s credentials
- :returns: A boolean if the cluster is running microk8s
+ :returns: A boolean if the cluster is running locally
"""
- for context in credentials['contexts']:
- if 'microk8s' in context['name']:
- return True
+ creds = yaml.safe_load(credentials)
+ if os.getenv("OSMLCM_VCA_APIPROXY"):
+ host_ip = os.getenv("OSMLCM_VCA_APIPROXY")
+
+ if creds and host_ip:
+ for cluster in creds['clusters']:
+ if 'server' in cluster['cluster']:
+ if host_ip in cluster['cluster']['server']:
+ return True
return False
- async def login(self):
+ async def login(self, cluster_uuid):
"""Login to the Juju controller."""
if self.authenticated:
self.connecting = True
+ # Test: Make sure we have the credentials loaded
+ config = self.get_config(cluster_uuid)
+
+ self.juju_endpoint = config['endpoint']
+ self.juju_user = config['username']
+ self.juju_secret = config['secret']
+ self.juju_ca_cert = config['cacert']
+ self.juju_public_key = None
+
self.controller = Controller()
if self.juju_secret:
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:
"""
# Remove the bootstrapped controller
- cmd = [self.k8scli, "remove-k8s", "--client", cloud_name]
- p = subprocess.run(
- cmd,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- encoding='ascii'
+ cmd = [self.juju_command, "remove-k8s", "--client", cloud_name]
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
)
- retcode = p.returncode
- if retcode > 0:
- raise Exception(p.stderr)
+ stdout, stderr = await process.communicate()
+
+ return_code = process.returncode
+
+ if return_code > 0:
+ raise Exception(stderr)
# Remove the cloud from the local config
- cmd = [self.k8scli, "remove-cloud", "--client", cloud_name]
- p = subprocess.run(
- cmd,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- encoding='ascii'
+ cmd = [self.juju_command, "remove-cloud", "--client", cloud_name]
+ process = await asyncio.create_subprocess_exec(
+ *cmd,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
)
- retcode = p.returncode
- if retcode > 0:
- raise Exception(p.stderr)
+ stdout, stderr = await process.communicate()
+
+ return_code = process.returncode
+ if return_code > 0:
+ raise Exception(stderr)
return True
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