X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FN2VC.git;a=blobdiff_plain;f=n2vc%2Fvnf.py;h=9441f4a646a190b6d3b97ae9a81a3a36bd47043c;hp=9cdbb330977a27fb41108409bd07bb0344f38cab;hb=ad3a05447418afef1d7b1998f9d7e351ccf9fe40;hpb=7bf2f4d5ba51d8a6909a8709aeda200ddb153b03;ds=sidebyside diff --git a/n2vc/vnf.py b/n2vc/vnf.py index 9cdbb33..9441f4a 100644 --- a/n2vc/vnf.py +++ b/n2vc/vnf.py @@ -1,4 +1,20 @@ +# Copyright 2019 Canonical Ltd. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + import asyncio +import base64 +import binascii import logging import os import os.path @@ -8,19 +24,23 @@ import ssl import subprocess import sys # import time +import n2vc.exceptions +from n2vc.provisioner import SSHProvisioner # FIXME: this should load the juju inside or modules without having to # explicitly install it. Check why it's not working. # Load our subtree of the juju library -path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) -path = os.path.join(path, "modules/libjuju/") -if path not in sys.path: - sys.path.insert(1, path) +# path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +# path = os.path.join(path, "modules/libjuju/") +# if path not in sys.path: +# sys.path.insert(1, path) +from juju.client import client from juju.controller import Controller from juju.model import ModelObserver from juju.errors import JujuAPIError, JujuError + # We might need this to connect to the websocket securely, but test and verify. try: ssl._create_default_https_context = ssl._create_unverified_context @@ -31,6 +51,7 @@ except AttributeError: # Custom exceptions +# Deprecated. Please use n2vc.exceptions namespace. class JujuCharmNotFound(Exception): """The Charm can't be found or is not readable.""" @@ -47,6 +68,10 @@ class NetworkServiceDoesNotExist(Exception): """The Network Service being acted against does not exist.""" +class PrimitiveDoesNotExist(Exception): + """The Primitive being executed does not exist.""" + + # Quiet the debug logging logging.getLogger('websockets.protocol').setLevel(logging.INFO) logging.getLogger('juju.client.connection').setLevel(logging.WARN) @@ -57,13 +82,12 @@ logging.getLogger('juju.machine').setLevel(logging.WARN) class VCAMonitor(ModelObserver): """Monitor state changes within the Juju Model.""" log = None - ns_name = None - applications = {} def __init__(self, ns_name): self.log = logging.getLogger(__name__) self.ns_name = ns_name + self.applications = {} def AddApplication(self, application_name, callback, *callback_args): if application_name not in self.applications: @@ -148,22 +172,40 @@ class N2VC: secret=None, artifacts=None, loop=None, + juju_public_key=None, + ca_cert=None, + api_proxy=None ): """Initialize N2VC - :param vcaconfig dict A dictionary containing the VCA configuration + Initializes the N2VC object, allowing the caller to interoperate with the VCA. + - :param artifacts str The directory where charms required by a vnfd are + :param log obj: The logging object to log to + :param server str: The IP Address or Hostname of the Juju controller + :param port int: The port of the Juju Controller + :param user str: The Juju username to authenticate with + :param secret str: The Juju password to authenticate with + :param artifacts str: The directory where charms required by a vnfd are stored. + :param loop obj: The loop to use. + :param juju_public_key str: The contents of the Juju public SSH key + :param ca_cert str: The CA certificate to use to authenticate + :param api_proxy str: The IP of the host machine :Example: - n2vc = N2VC(vcaconfig={ - 'secret': 'MzI3MDJhOTYxYmM0YzRjNTJiYmY1Yzdm', - 'user': 'admin', - 'ip-address': '10.44.127.137', - 'port': 17070, - 'artifacts': '/path/to/charms' - }) + client = n2vc.vnf.N2VC( + log=log, + server='10.1.1.28', + port=17070, + user='admin', + secret='admin', + artifacts='/app/storage/myvnf/charms', + loop=loop, + juju_public_key='', + ca_cert='', + api_proxy='192.168.1.155' + ) """ # Initialize instance-level variables @@ -172,6 +214,12 @@ class N2VC: self.controller = None self.connecting = False self.authenticated = False + self.api_proxy = api_proxy + + if log: + self.log = log + else: + self.log = logging.getLogger(__name__) # For debugging self.refcount = { @@ -190,10 +238,38 @@ class N2VC: self.username = "" self.secret = "" - if log: - self.log = log + self.juju_public_key = juju_public_key + if juju_public_key: + self._create_juju_public_key(juju_public_key) else: - self.log = logging.getLogger(__name__) + self.juju_public_key = '' + + # TODO: Verify ca_cert is valid before using. VCA will crash + # if the ca_cert isn't formatted correctly. + def base64_to_cacert(b64string): + """Convert the base64-encoded string containing the VCA CACERT. + + The input string.... + + """ + try: + cacert = base64.b64decode(b64string).decode("utf-8") + + cacert = re.sub( + r'\\n', + r'\n', + cacert, + ) + except binascii.Error as e: + self.log.debug("Caught binascii.Error: {}".format(e)) + raise n2vc.exceptions.InvalidCACertificate("Invalid CA Certificate") + + return cacert + + self.ca_cert = None + if ca_cert: + self.ca_cert = base64_to_cacert(ca_cert) + # Quiet websocket traffic logging.getLogger('websockets.protocol').setLevel(logging.INFO) @@ -222,6 +298,31 @@ class N2VC: """Close any open connections.""" yield self.logout() + def _create_juju_public_key(self, public_key): + """Recreate the Juju public key on disk. + + Certain libjuju commands expect to be run from the same machine as Juju + is bootstrapped to. This method will write the public key to disk in + that location: ~/.local/share/juju/ssh/juju_id_rsa.pub + """ + # Make sure that we have a public key before writing to disk + if public_key is None or len(public_key) == 0: + if 'OSM_VCA_PUBKEY' in os.environ: + public_key = os.getenv('OSM_VCA_PUBKEY', '') + if len(public_key == 0): + return + else: + return + + path = "{}/.local/share/juju/ssh".format( + os.path.expanduser('~'), + ) + if not os.path.exists(path): + os.makedirs(path) + + with open('{}/juju_id_rsa.pub'.format(path), 'w') as f: + f.write(public_key) + def notify_callback(self, model_name, application_name, status, message, callback=None, *callback_args): try: @@ -245,9 +346,10 @@ class N2VC: vdu: ... - relation: - - provides: dataVM:db - requires: mgmtVM:app + vca-relationships: + relation: + - provides: dataVM:db + requires: mgmtVM:app This tells N2VC that the charm referred to by the dataVM vdu offers a relation named 'db', and the mgmtVM vdu has an 'app' endpoint that should be connected to a database. @@ -297,8 +399,9 @@ class N2VC: # Loop through relations for cfg in configs: if 'juju' in cfg: - if 'relation' in juju: - for rel in juju['relation']: + juju = cfg['juju'] + if 'vca-relationships' in juju and 'relation' in juju['vca-relationships']: + for rel in juju['vca-relationships']['relation']: try: # get the application name for the provides @@ -372,6 +475,7 @@ class N2VC: self.notify_callback( model_name, application_name, + "error", "failed", callback, *callback_args, @@ -401,25 +505,10 @@ class N2VC: # Register this application with the model-level event monitor # ################################################################ if callback: - self.monitors[model_name].AddApplication( + self.log.debug("JujuApi: Registering callback for {}".format( application_name, - callback, - *callback_args - ) - - ######################################################## - # Check for specific machine placement (native charms) # - ######################################################## - to = "" - if machine_spec.keys(): - if all(k in machine_spec for k in ['host', 'user']): - # Enlist an existing machine as a Juju unit - machine = await model.add_machine(spec='ssh:{}@{}:{}'.format( - machine_spec['user'], - machine_spec['host'], - self.GetPrivateKeyPath(), - )) - to = machine.id + )) + await self.Subscribe(model_name, application_name, callback, *callback_args) ####################################### # Get the initial charm configuration # @@ -437,11 +526,73 @@ class N2VC: {'': rw_mgmt_ip} ) - self.log.debug("JujuApi: Deploying charm ({}/{}) from {}".format( + ######################################################## + # Check for specific machine placement (native charms) # + ######################################################## + to = "" + series = "xenial" + + if machine_spec.keys(): + if all(k in machine_spec for k in ['hostname', 'username']): + + # Allow series to be derived from the native charm + series = None + + self.log.debug("Provisioning manual machine {}@{}".format( + machine_spec['username'], + machine_spec['hostname'], + )) + + """Native Charm support + + Taking a bare VM (assumed to be an Ubuntu cloud image), + the provisioning process will: + - Create an ubuntu user w/sudo access + - Detect hardware + - Detect architecture + - Download and install Juju agent from controller + - Enable Juju agent + - Add an iptables rule to route traffic to the API proxy + """ + + to = await self.provision_machine( + model_name=model_name, + username=machine_spec['username'], + hostname=machine_spec['hostname'], + private_key_path=self.GetPrivateKeyPath(), + ) + self.log.debug("Provisioned machine id {}".format(to)) + + # TODO: If to is none, raise an exception + + # The native charm won't have the sshproxy layer, typically, but LCM uses the config primitive + # to interpret what the values are. That's a gap to fill. + + """ + The ssh-* config parameters are unique to the sshproxy layer, + which most native charms will not be aware of. + + Setting invalid config parameters will cause the deployment to + fail. + + For the moment, we will strip the ssh-* parameters from native + charms, until the feature gap is addressed in the information + model. + """ + + # Native charms don't include the ssh-* config values, so strip them + # from the initial_config, otherwise the deploy will raise an error. + # self.log.debug("Removing ssh-* from initial-config") + for k in ['ssh-hostname', 'ssh-username', 'ssh-password']: + if k in initial_config: + self.log.debug("Removing parameter {}".format(k)) + del initial_config[k] + + self.log.debug("JujuApi: Deploying charm ({}/{}) from {} to {}".format( model_name, application_name, charm_path, - to=to, + to, )) ######################################################## @@ -455,16 +606,26 @@ class N2VC: application_name=application_name, # Proxy charms should use the current LTS. This will need to be # changed for native charms. - series='xenial', + series=series, # Apply the initial 'config' primitive during deployment config=initial_config, # Where to deploy the charm to. to=to, ) - # Map the vdu id<->app name, - # - await self.Relate(model_name, vnfd) + ############################# + # Map the vdu id<->app name # + ############################# + try: + await self.Relate(model_name, vnfd) + except KeyError as ex: + # We don't currently support relations between NS and VNF/VDU charms + self.log.warn("[N2VC] Relations not supported: {}".format(ex)) + except Exception as ex: + # This may happen if not all of the charms needed by the relation + # are ready. We can safely ignore this, because Relate will be + # retried when the endpoint of the relation is deployed. + self.log.warn("[N2VC] Relations not ready") # ####################################### # # Execute initial config primitive(s) # @@ -671,16 +832,25 @@ class N2VC: } for primitive in sorted(primitives): - uuids.append( - await self.ExecutePrimitive( - model_name, - application_name, - primitives[primitive]['name'], - callback, - callback_args, - **primitives[primitive]['parameters'], + try: + # self.log.debug("Queuing action {}".format(primitives[primitive]['name'])) + uuids.append( + await self.ExecutePrimitive( + model_name, + application_name, + primitives[primitive]['name'], + callback, + callback_args, + **primitives[primitive]['parameters'], + ) ) - ) + except PrimitiveDoesNotExist as e: + self.log.debug("Ignoring exception PrimitiveDoesNotExist: {}".format(e)) + pass + except Exception as e: + self.log.debug("XXXXXXXXXXXXXXXXXXXXXXXXX Unexpected exception: {}".format(e)) + raise e + except N2VCPrimitiveExecutionFailed as e: self.log.debug( "[N2VC] Exception executing primitive: {}".format(e) @@ -727,12 +897,22 @@ class N2VC: else: app = await self.get_application(model, application_name) if app: + # Does this primitive exist? + actions = await app.get_actions() + + if primitive not in actions.keys(): + raise PrimitiveDoesNotExist("Primitive {} does not exist".format(primitive)) + # Run against the first (and probably only) unit in the app unit = app.units[0] if unit: action = await unit.run_action(primitive, **params) uuid = action.id + except PrimitiveDoesNotExist as e: + # Catch and raise this exception if it's thrown from the inner block + raise e except Exception as e: + # An unexpected exception was caught self.log.debug( "Caught exception while executing primitive: {}".format(e) ) @@ -759,7 +939,7 @@ class N2VC: app = await self.get_application(model, application_name) if app: # Remove this application from event monitoring - self.monitors[model_name].RemoveApplication(application_name) + await self.Unsubscribe(model_name, application_name) # self.notify_callback(model_name, application_name, "removing", callback, *callback_args) self.log.debug( @@ -767,13 +947,13 @@ class N2VC: ) await app.remove() - await self.disconnect_model(self.monitors[model_name]) + # await self.disconnect_model(self.monitors[model_name]) - # Notify the callback that this charm has been removed. self.notify_callback( model_name, application_name, "removed", + "Removing charm {}".format(application_name), callback, *callback_args, ) @@ -798,16 +978,8 @@ class N2VC: models = await self.controller.list_models() if ns_uuid not in models: - try: - self.models[ns_uuid] = await self.controller.add_model( - ns_uuid - ) - except JujuError as e: - if "already exists" not in e.message: - raise e - - # Create an observer for this model - await self.create_model_monitor(ns_uuid) + # Get the new model + await self.get_model(ns_uuid) return True @@ -823,20 +995,59 @@ class N2VC: # Do not delete the default model. The default model was used by all # Network Services, prior to the implementation of a model per NS. - if ns_uuid.lower() is "default": + if ns_uuid.lower() == "default": return False if not self.authenticated: - self.log.debug("Authenticating with Juju") await self.login() + models = await self.controller.list_models() + if ns_uuid in models: + model = await self.controller.get_model(ns_uuid) + + for application in model.applications: + app = model.applications[application] + + await self.RemoveCharms(ns_uuid, application) + + self.log.debug("Unsubscribing Watcher for {}".format(application)) + await self.Unsubscribe(ns_uuid, application) + + self.log.debug("Waiting for application to terminate") + timeout = 30 + try: + await model.block_until( + lambda: all( + unit.workload_status in ['terminated'] for unit in app.units + ), + timeout=timeout + ) + except Exception as e: + self.log.debug("Timed out waiting for {} to terminate.".format(application)) + + for machine in model.machines: + try: + self.log.debug("Destroying machine {}".format(machine)) + await model.machines[machine].destroy(force=True) + except JujuAPIError as e: + if 'does not exist' in str(e): + # Our cached model may be stale, because the machine + # has already been removed. It's safe to continue. + continue + else: + self.log.debug("Caught exception: {}".format(e)) + raise e + # Disconnect from the Model if ns_uuid in self.models: - await self.disconnect_model(self.models[ns_uuid]) + self.log.debug("Disconnecting model {}".format(ns_uuid)) + # await self.disconnect_model(self.models[ns_uuid]) + await self.disconnect_model(ns_uuid) try: + self.log.debug("Destroying model {}".format(ns_uuid)) await self.controller.destroy_models(ns_uuid) - except JujuError as e: + except JujuError: raise NetworkServiceDoesNotExist( "The Network Service '{}' does not exist".format(ns_uuid) ) @@ -864,6 +1075,33 @@ class N2VC: return True return False + async def Subscribe(self, ns_name, application_name, callback, *callback_args): + """Subscribe to callbacks for an application. + + :param ns_name str: The name of the Network Service + :param application_name str: The name of the application + :param callback obj: The callback method + :param callback_args list: The list of arguments to append to calls to + the callback method + """ + self.monitors[ns_name].AddApplication( + application_name, + callback, + *callback_args + ) + + async def Unsubscribe(self, ns_name, application_name): + """Unsubscribe to callbacks for an application. + + Unsubscribes the caller from notifications from a deployed application. + + :param ns_name str: The name of the Network Service + :param application_name str: The name of the application + """ + self.monitors[ns_name].RemoveApplication( + application_name, + ) + # Non-public methods async def add_relation(self, model_name, relation1, relation2): """ @@ -1052,7 +1290,9 @@ class N2VC: if model_name not in models: try: self.models[model_name] = await self.controller.add_model( - model_name + model_name, + config={'authorized-keys': self.juju_public_key} + ) except JujuError as e: if "already exists" not in e.message: @@ -1090,25 +1330,28 @@ class N2VC: self.log.debug("JujuApi: Logging into controller") - cacert = None self.controller = Controller(loop=self.loop) if self.secret: self.log.debug( - "Connecting to controller... ws://{}:{} as {}/{}".format( + "Connecting to controller... ws://{} as {}/{}".format( self.endpoint, - self.port, self.user, self.secret, ) ) - await self.controller.connect( - endpoint=self.endpoint, - username=self.user, - password=self.secret, - cacert=cacert, - ) - self.refcount['controller'] += 1 + try: + await self.controller.connect( + endpoint=self.endpoint, + username=self.user, + password=self.secret, + cacert=self.ca_cert, + ) + self.refcount['controller'] += 1 + self.authenticated = True + self.log.debug("JujuApi: Logged into controller") + except Exception as ex: + self.log.debug("Caught exception: {}".format(ex)) else: # current_controller no longer exists # self.log.debug("Connecting to current controller...") @@ -1119,9 +1362,8 @@ class N2VC: # cacert=cacert, # ) self.log.fatal("VCA credentials not configured.") + self.authenticated = False - self.authenticated = True - self.log.debug("JujuApi: Logged into controller") async def logout(self): """Logout of the Juju controller.""" @@ -1154,10 +1396,81 @@ class N2VC: async def disconnect_model(self, model): self.log.debug("Disconnecting model {}".format(model)) if model in self.models: - print("Disconnecting model") - await self.models[model].disconnect() - self.refcount['model'] -= 1 - self.models[model] = None + try: + await self.models[model].disconnect() + self.refcount['model'] -= 1 + self.models[model] = None + except Exception as e: + self.log.debug("Caught exception: {}".format(e)) + + async def provision_machine(self, model_name: str, + hostname: str, username: str, + private_key_path: str) -> int: + """Provision a machine. + + This executes the SSH provisioner, which will log in to a machine via + SSH and prepare it for use with the Juju model + + :param model_name str: The name of the model + :param hostname str: The IP or hostname of the target VM + :param user str: The username to login to + :param private_key_path str: The path to the private key that's been injected to the VM via cloud-init + :return machine_id int: Returns the id of the machine or None if provisioning fails + """ + if not self.authenticated: + await self.login() + + machine_id = None + + if self.api_proxy: + self.log.debug("Instantiating SSH Provisioner for {}@{} ({})".format( + username, + hostname, + private_key_path + )) + provisioner = SSHProvisioner( + host=hostname, + user=username, + private_key_path=private_key_path, + log=self.log, + ) + + params = None + try: + params = provisioner.provision_machine() + except Exception as ex: + self.log.debug("caught exception from provision_machine: {}".format(ex)) + return None + + if params: + params.jobs = ['JobHostUnits'] + + model = await self.get_model(model_name) + + connection = model.connection() + + # Submit the request. + self.log.debug("Adding machine to model") + client_facade = client.ClientFacade.from_connection(connection) + results = await client_facade.AddMachines(params=[params]) + error = results.machines[0].error + if error: + raise ValueError("Error adding machine: %s" % error.message) + + machine_id = results.machines[0].machine + + # Need to run this after AddMachines has been called, + # as we need the machine_id + self.log.debug("Installing Juju agent") + await provisioner.install_agent( + connection, + params.nonce, + machine_id, + self.api_proxy, + ) + else: + self.log.debug("Missing API Proxy") + return machine_id # async def remove_application(self, name): # """Remove the application."""