-
+import asyncio
import logging
import os
import os.path
import re
+import shlex
import ssl
+import subprocess
import sys
-import time
+# import time
# FIXME: this should load the juju inside or modules without having to
# explicitly install it. Check why it's not working.
sys.path.insert(1, path)
from juju.controller import Controller
-from juju.model import Model, ModelObserver
+from juju.model import ModelObserver
# We might need this to connect to the websocket securely, but test and verify.
class JujuApplicationExists(Exception):
"""The Application already exists."""
+
class N2VCPrimitiveExecutionFailed(Exception):
"""Something failed while attempting to execute a primitive."""
logging.getLogger('juju.model').setLevel(logging.WARN)
logging.getLogger('juju.machine').setLevel(logging.WARN)
+
class VCAMonitor(ModelObserver):
"""Monitor state changes within the Juju Model."""
log = None
application_name = delta.data['application']
callback = self.applications[application_name]['callback']
- callback_args = self.applications[application_name]['callback_args']
+ callback_args = \
+ self.applications[application_name]['callback_args']
if old and new:
- old_status = old.workload_status
- new_status = new.workload_status
-
- if old_status == new_status:
- """The workload status may fluctuate around certain events,
- so wait until the status has stabilized before triggering
- the callback."""
- if callback:
- callback(
- self.ns_name,
- delta.data['application'],
- new_status,
- *callback_args)
+ # Fire off a callback with the application state
+ if callback:
+ callback(
+ self.ns_name,
+ delta.data['application'],
+ new.workload_status,
+ new.workload_status_message,
+ *callback_args)
if old and not new:
# This is a charm being removed
self.ns_name,
delta.data['application'],
"removed",
+ "",
*callback_args)
except Exception as e:
- self.log.debug("[1] notify_callback exception {}".format(e))
+ self.log.debug("[1] notify_callback exception: {}".format(e))
+
elif delta.entity == "action":
# TODO: Decide how we want to notify the user of actions
# Create unique models per network service
# Document all public functions
-class N2VC:
-
- # Juju API
- api = None
- log = None
- controller = None
- connecting = False
- authenticated = False
-
- models = {}
- default_model = None
-
- # Model Observers
- monitors = {}
-
- # VCA config
- hostname = ""
- port = 17070
- username = ""
- secret = ""
+class N2VC:
def __init__(self,
log=None,
server='127.0.0.1',
port=17070,
user='admin',
secret=None,
- artifacts=None
+ artifacts=None,
+ loop=None,
):
"""Initialize N2VC
'port': 17070,
'artifacts': '/path/to/charms'
})
-
"""
+ # Initialize instance-level variables
+ self.api = None
+ self.log = None
+ self.controller = None
+ self.connecting = False
+ self.authenticated = False
+
+ # For debugging
+ self.refcount = {
+ 'controller': 0,
+ 'model': 0,
+ }
+
+ self.models = {}
+ self.default_model = None
+
+ # Model Observers
+ self.monitors = {}
+
+ # VCA config
+ self.hostname = ""
+ self.port = 17070
+ self.username = ""
+ self.secret = ""
+
if log:
self.log = log
else:
self.artifacts = artifacts
+ self.loop = loop or asyncio.get_event_loop()
+
def __del__(self):
"""Close any open connections."""
yield self.logout()
- def notify_callback(self, model_name, application_name, status, callback=None, *callback_args):
+ def notify_callback(self, model_name, application_name, status, message,
+ callback=None, *callback_args):
try:
if callback:
- callback(model_name, application_name, status, *callback_args)
+ callback(
+ model_name,
+ application_name,
+ status, message,
+ *callback_args,
+ )
except Exception as e:
self.log.error("[0] notify_callback exception {}".format(e))
raise e
return self.default_model
- async def DeployCharms(self, model_name, application_name, vnfd, charm_path, params={}, machine_spec={}, callback=None, *callback_args):
+ async def DeployCharms(self, model_name, application_name, vnfd,
+ charm_path, params={}, machine_spec={},
+ callback=None, *callback_args):
"""Deploy one or more charms associated with a VNF.
Deploy the charm(s) referenced in a VNF Descriptor.
- You can pass either the nsd record or the id of the network
- service, but this method will fail without one of them.
-
- :param str ns_name: The name of the network service
+ :param str model_name: The name of the network service.
:param str application_name: The name of the application
:param dict vnfd: The name of the application
:param str charm_path: The path to the Juju charm
# Pass the initial-config-primitives section of the vnf or vdu
'initial-config-primitives': {...}
}
- :param dict machine_spec: A dictionary describing the machine to install to
+ :param dict machine_spec: A dictionary describing the machine to
+ install to
Examples::
{
'hostname': '1.2.3.4',
'username': 'ubuntu',
}
:param obj callback: A callback function to receive status changes.
- :param tuple callback_args: A list of arguments to be passed to the callback
+ :param tuple callback_args: A list of arguments to be passed to the
+ callback
"""
########################################################
########################################################
if not os.path.exists(charm_path):
self.log.debug("Charm path doesn't exist: {}".format(charm_path))
- self.notify_callback(model_name, application_name, "failed", callback, *callback_args)
+ self.notify_callback(
+ model_name,
+ application_name,
+ "failed",
+ callback,
+ *callback_args,
+ )
raise JujuCharmNotFound("No artifacts configured.")
################################
########################################
app = await self.get_application(model, application_name)
if app:
- raise JujuApplicationExists("Can't deploy application \"{}\" to model \"{}\" because it already exists.".format(application_name, model))
+ raise JujuApplicationExists("Can't deploy application \"{}\" to model \"{}\" because it already exists.".format(application_name, model_name))
################################################################
# Register this application with the model-level event monitor #
########################################################
to = ""
if machine_spec.keys():
- # TODO: This needs to be tested.
- # if all(k in machine_spec for k in ['hostname', 'username']):
- # # Enlist the existing machine in Juju
- # machine = await self.model.add_machine(spec='ssh:%@%'.format(
- # specs['host'],
- # specs['user'],
- # ))
- # to = machine.id
- pass
+ 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
#######################################
# Get the initial charm configuration #
if 'rw_mgmt_ip' in params:
rw_mgmt_ip = params['rw_mgmt_ip']
+ if 'initial-config-primitive' not in params:
+ params['initial-config-primitive'] = {}
+
initial_config = self._get_config_from_dict(
params['initial-config-primitive'],
{'<rw_mgmt_ip>': rw_mgmt_ip}
series='xenial',
# Apply the initial 'config' primitive during deployment
config=initial_config,
- # TBD: Where to deploy the charm to.
- to=None,
+ # Where to deploy the charm to.
+ to=to,
)
# #######################################
# # Execute initial config primitive(s) #
# #######################################
+ await self.ExecuteInitialPrimitives(
+ model_name,
+ application_name,
+ params,
+ )
+
+ # primitives = {}
+ #
+ # # Build a sequential list of the primitives to execute
+ # for primitive in params['initial-config-primitive']:
+ # try:
+ # if primitive['name'] == 'config':
+ # # This is applied when the Application is deployed
+ # pass
+ # else:
+ # seq = primitive['seq']
+ #
+ # params = {}
+ # if 'parameter' in primitive:
+ # params = primitive['parameter']
+ #
+ # primitives[seq] = {
+ # 'name': primitive['name'],
+ # 'parameters': self._map_primitive_parameters(
+ # params,
+ # {'<rw_mgmt_ip>': rw_mgmt_ip}
+ # ),
+ # }
+ #
+ # for primitive in sorted(primitives):
+ # await self.ExecutePrimitive(
+ # model_name,
+ # application_name,
+ # primitives[primitive]['name'],
+ # callback,
+ # callback_args,
+ # **primitives[primitive]['parameters'],
+ # )
+ # except N2VCPrimitiveExecutionFailed as e:
+ # self.log.debug(
+ # "[N2VC] Exception executing primitive: {}".format(e)
+ # )
+ # raise
+
+ async def GetPrimitiveStatus(self, model_name, uuid):
+ """Get the status of an executed Primitive.
+
+ The status of an executed Primitive will be one of three values:
+ - completed
+ - failed
+ - running
+ """
+ status = None
+ try:
+ if not self.authenticated:
+ await self.login()
+
+ # FIXME: This is hard-coded until model-per-ns is added
+ model_name = 'default'
+
+ model = await self.get_model(model_name)
+
+ results = await model.get_action_status(uuid)
+
+ if uuid in results:
+ status = results[uuid]
+
+ except Exception as e:
+ self.log.debug(
+ "Caught exception while getting primitive status: {}".format(e)
+ )
+ raise N2VCPrimitiveExecutionFailed(e)
+
+ return status
+
+ async def GetPrimitiveOutput(self, model_name, uuid):
+ """Get the output of an executed Primitive.
+
+ Note: this only returns output for a successfully executed primitive.
+ """
+ results = None
+ try:
+ if not self.authenticated:
+ await self.login()
+
+ # FIXME: This is hard-coded until model-per-ns is added
+ model_name = 'default'
+
+ model = await self.get_model(model_name)
+ results = await model.get_action_output(uuid, 60)
+ except Exception as e:
+ self.log.debug(
+ "Caught exception while getting primitive status: {}".format(e)
+ )
+ raise N2VCPrimitiveExecutionFailed(e)
+
+ return results
+
+ # async def ProvisionMachine(self, model_name, hostname, username):
+ # """Provision machine for usage with Juju.
+ #
+ # Provisions a previously instantiated machine for use with Juju.
+ # """
+ # try:
+ # if not self.authenticated:
+ # await self.login()
+ #
+ # # FIXME: This is hard-coded until model-per-ns is added
+ # model_name = 'default'
+ #
+ # model = await self.get_model(model_name)
+ # model.add_machine(spec={})
+ #
+ # machine = await model.add_machine(spec='ssh:{}@{}:{}'.format(
+ # "ubuntu",
+ # host['address'],
+ # private_key_path,
+ # ))
+ # return machine.id
+ #
+ # except Exception as e:
+ # self.log.debug(
+ # "Caught exception while getting primitive status: {}".format(e)
+ # )
+ # raise N2VCPrimitiveExecutionFailed(e)
+
+ def GetPrivateKeyPath(self):
+ homedir = os.environ['HOME']
+ sshdir = "{}/.ssh".format(homedir)
+ private_key_path = "{}/id_n2vc_rsa".format(sshdir)
+ return private_key_path
+
+ async def GetPublicKey(self):
+ """Get the N2VC SSH public key.abs
+
+ Returns the SSH public key, to be injected into virtual machines to
+ be managed by the VCA.
+
+ The first time this is run, a ssh keypair will be created. The public
+ key is injected into a VM so that we can provision the machine with
+ Juju, after which Juju will communicate with the VM directly via the
+ juju agent.
+ """
+ public_key = ""
+
+ # Find the path to where we expect our key to live.
+ homedir = os.environ['HOME']
+ sshdir = "{}/.ssh".format(homedir)
+ if not os.path.exists(sshdir):
+ os.mkdir(sshdir)
+
+ private_key_path = "{}/id_n2vc_rsa".format(sshdir)
+ public_key_path = "{}.pub".format(private_key_path)
+
+ # If we don't have a key generated, generate it.
+ if not os.path.exists(private_key_path):
+ cmd = "ssh-keygen -t {} -b {} -N '' -f {}".format(
+ "rsa",
+ "4096",
+ private_key_path
+ )
+ subprocess.check_output(shlex.split(cmd))
+
+ # Read the public key
+ with open(public_key_path, "r") as f:
+ public_key = f.readline()
+
+ return public_key
+
+ async def ExecuteInitialPrimitives(self, model_name, application_name,
+ params, callback=None, *callback_args):
+ """Execute multiple primitives.
+
+ Execute multiple primitives as declared in initial-config-primitive.
+ This is useful in cases where the primitives initially failed -- for
+ example, if the charm is a proxy but the proxy hasn't been configured
+ yet.
+ """
+ uuids = []
primitives = {}
# Build a sequential list of the primitives to execute
for primitive in params['initial-config-primitive']:
try:
if primitive['name'] == 'config':
- # This is applied when the Application is deployed
pass
else:
seq = primitive['seq']
+ params = {}
+ if 'parameter' in primitive:
+ params = primitive['parameter']
+
primitives[seq] = {
'name': primitive['name'],
'parameters': self._map_primitive_parameters(
- primitive['parameter'],
- {'<rw_mgmt_ip>': rw_mgmt_ip}
+ params,
+ {'<rw_mgmt_ip>': None}
),
}
for primitive in sorted(primitives):
- await self.ExecutePrimitive(
- model_name,
- application_name,
- primitives[primitive]['name'],
- callback,
- callback_args,
- **primitives[primitive]['parameters'],
+ uuids.append(
+ await self.ExecutePrimitive(
+ model_name,
+ application_name,
+ primitives[primitive]['name'],
+ callback,
+ callback_args,
+ **primitives[primitive]['parameters'],
+ )
)
except N2VCPrimitiveExecutionFailed as e:
- self.debug.log(
+ self.log.debug(
"[N2VC] Exception executing primitive: {}".format(e)
)
raise
+ return uuids
+
+ async def ExecutePrimitive(self, model_name, application_name, primitive,
+ callback, *callback_args, **params):
+ """Execute a primitive of a charm for Day 1 or Day 2 configuration.
+
+ Execute a primitive defined in the VNF descriptor.
- async def ExecutePrimitive(self, model_name, application_name, primitive, callback, *callback_args, **params):
+ :param str model_name: The name of the network service.
+ :param str application_name: The name of the application
+ :param str primitive: The name of the primitive to execute.
+ :param obj callback: A callback function to receive status changes.
+ :param tuple callback_args: A list of arguments to be passed to the
+ callback function.
+ :param dict params: A dictionary of key=value pairs representing the
+ primitive's parameters
+ Examples::
+ {
+ 'rw_mgmt_ip': '1.2.3.4',
+ # Pass the initial-config-primitives section of the vnf or vdu
+ 'initial-config-primitives': {...}
+ }
+ """
+ self.log.debug("Executing {}".format(primitive))
+ uuid = None
try:
if not self.authenticated:
await self.login()
# FIXME: This is hard-coded until model-per-ns is added
model_name = 'default'
+ model = await self.get_model(model_name)
+
if primitive == 'config':
# config is special, and expecting params to be a dictionary
- await self.set_config(application_name, params['params'])
+ await self.set_config(
+ model,
+ application_name,
+ params['params'],
+ )
else:
- model = await self.controller.get_model(model_name)
app = await self.get_application(model, application_name)
if app:
# Run against the first (and probably only) unit in the app
unit = app.units[0]
if unit:
- self.log.debug("Executing primitive {}".format(primitive))
action = await unit.run_action(primitive, **params)
- # action = await action.wait()
- await model.disconnect()
+ uuid = action.id
except Exception as e:
- self.log.debug("Caught exception while executing primitive: {}".format(e))
- raise e
+ self.log.debug(
+ "Caught exception while executing primitive: {}".format(e)
+ )
+ raise N2VCPrimitiveExecutionFailed(e)
+ return uuid
- async def RemoveCharms(self, model_name, application_name, callback=None, *callback_args):
+ async def RemoveCharms(self, model_name, application_name, callback=None,
+ *callback_args):
+ """Remove a charm from the VCA.
+
+ Remove a charm referenced in a VNF Descriptor.
+
+ :param str model_name: The name of the network service.
+ :param str application_name: The name of the application
+ :param obj callback: A callback function to receive status changes.
+ :param tuple callback_args: A list of arguments to be passed to the
+ callback function.
+ """
try:
if not self.authenticated:
await self.login()
self.monitors[model_name].RemoveApplication(application_name)
# self.notify_callback(model_name, application_name, "removing", callback, *callback_args)
- self.log.debug("Removing the application {}".format(application_name))
+ self.log.debug(
+ "Removing the application {}".format(application_name)
+ )
await app.remove()
# Notify the callback that this charm has been removed.
- self.notify_callback(model_name, application_name, "removed", callback, *callback_args)
+ self.notify_callback(
+ model_name,
+ application_name,
+ "removed",
+ callback,
+ *callback_args,
+ )
except Exception as e:
print("Caught exception: {}".format(e))
async def DestroyNetworkService(self, nsd):
raise NotImplementedError()
- async def GetMetrics(self, nsd, vnfd):
- """Get the metrics collected by the VCA."""
- raise NotImplementedError()
+ async def GetMetrics(self, model_name, application_name):
+ """Get the metrics collected by the VCA.
+
+ :param model_name The name of the model
+ :param application_name The name of the application
+ """
+ metrics = {}
+ model = await self.get_model(model_name)
+ app = await self.get_application(model, application_name)
+ if app:
+ metrics = await app.get_metrics()
+
+ return metrics
+
+ async def HasApplication(self, model_name, application_name):
+ model = await self.get_model(model_name)
+ app = await self.get_application(model, application_name)
+ if app:
+ return True
+ return False
# Non-public methods
async def add_relation(self, a, b, via=None):
finally:
await m.disconnect()
- async def apply_config(self, config, application):
- """Apply a configuration to the application."""
- print("JujuApi: Applying configuration to {}.".format(
- application
- ))
- return await self.set_config(application=application, config=config)
+ # async def apply_config(self, config, application):
+ # """Apply a configuration to the application."""
+ # print("JujuApi: Applying configuration to {}.".format(
+ # application
+ # ))
+ # return await self.set_config(application=application, config=config)
def _get_config_from_dict(self, config_primitive, values):
"""Transform the yang config primitive to dict.
params = {}
for parameter in parameters:
param = str(parameter['name'])
+
+ # Typecast parameter value, if present
+ if 'data-type' in parameter:
+ paramtype = str(parameter['data-type']).lower()
+ value = None
+
+ if paramtype == "integer":
+ value = int(parameter['value'])
+ elif paramtype == "boolean":
+ value = bool(parameter['value'])
+ else:
+ value = str(parameter['value'])
+
if parameter['value'] == "<rw_mgmt_ip>":
params[param] = str(values[parameter['value']])
else:
- params[param] = str(parameter['value'])
+ params[param] = value
return params
def _get_config_from_yang(self, config_primitive, values):
return config
+ @staticmethod
def FormatApplicationName(self, *args):
"""
Generate a Juju-compatible Application name
appname += c
return re.sub('\-+', '-', appname.lower())
-
# def format_application_name(self, nsd_name, vnfr_name, member_vnf_index=0):
# """Format the name of the application
#
await self.login()
if model_name not in self.models:
- print("connecting to model {}".format(model_name))
- self.models[model_name] = await self.controller.get_model(model_name)
+ self.models[model_name] = await self.controller.get_model(
+ model_name,
+ )
+ self.refcount['model'] += 1
# Create an observer for this model
self.monitors[model_name] = VCAMonitor(model_name)
self.log.debug("JujuApi: Logging into controller")
cacert = None
- self.controller = Controller()
+ self.controller = Controller(loop=self.loop)
if self.secret:
- self.log.debug("Connecting to controller... ws://{}:{} as {}/{}".format(self.endpoint, self.port, self.user, self.secret))
+ self.log.debug(
+ "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
else:
# current_controller no longer exists
# self.log.debug("Connecting to current controller...")
self.authenticated = True
self.log.debug("JujuApi: Logged into controller")
- # self.default_model = await self.controller.get_model("default")
-
async def logout(self):
"""Logout of the Juju controller."""
if not self.authenticated:
try:
if self.default_model:
- self.log.debug("Disconnecting model {}".format(self.default_model))
+ self.log.debug("Disconnecting model {}".format(
+ self.default_model
+ ))
await self.default_model.disconnect()
+ self.refcount['model'] -= 1
self.default_model = None
for model in self.models:
await self.models[model].disconnect()
+ self.refcount['model'] -= 1
+ self.models[model] = None
if self.controller:
- self.log.debug("Disconnecting controller {}".format(self.controller))
+ self.log.debug("Disconnecting controller {}".format(
+ self.controller
+ ))
await self.controller.disconnect()
- # self.controller = None
+ self.refcount['controller'] -= 1
+ self.controller = None
self.authenticated = False
+
+ self.log.debug(self.refcount)
+
except Exception as e:
- self.log.fail("Fatal error logging out of Juju Controller: {}".format(e))
+ self.log.fatal(
+ "Fatal error logging out of Juju Controller: {}".format(e)
+ )
raise e
-
# async def remove_application(self, name):
# """Remove the application."""
# if not self.authenticated:
app = await self.get_application(self.default_model, application)
if app:
- self.log.debug("JujuApi: Resolving errors for application {}".format(
- application,
- ))
+ self.log.debug(
+ "JujuApi: Resolving errors for application {}".format(
+ application,
+ )
+ )
for unit in app.units:
app.resolved(retry=True)
# so use the first unit available.
unit = app.units[0]
- self.log.debug("JujuApi: Running Action {} against Application {}".format(
- action_name,
- application,
- ))
+ self.log.debug(
+ "JujuApi: Running Action {} against Application {}".format(
+ action_name,
+ application,
+ )
+ )
action = await unit.run_action(action_name, **params)
return result
- async def set_config(self, application, config):
+ async def set_config(self, model_name, application, config):
"""Apply a configuration to the application."""
if not self.authenticated:
await self.login()
- app = await self.get_application(self.default_model, application)
+ app = await self.get_application(model_name, application)
if app:
self.log.debug("JujuApi: Setting config for Application {}".format(
application,
if config[key] != newconf[key]['value']:
self.log.debug("JujuApi: Config not set! Key {} Value {} doesn't match {}".format(key, config[key], newconf[key]))
- async def set_parameter(self, parameter, value, application=None):
- """Set a config parameter for a service."""
- if not self.authenticated:
- await self.login()
-
- self.log.debug("JujuApi: Setting {}={} for Application {}".format(
- parameter,
- value,
- application,
- ))
- return await self.apply_config(
- {parameter: value},
- application=application,
- )
-
- async def wait_for_application(self, name, timeout=300):
+ # async def set_parameter(self, parameter, value, application=None):
+ # """Set a config parameter for a service."""
+ # if not self.authenticated:
+ # await self.login()
+ #
+ # self.log.debug("JujuApi: Setting {}={} for Application {}".format(
+ # parameter,
+ # value,
+ # application,
+ # ))
+ # return await self.apply_config(
+ # {parameter: value},
+ # application=application,
+ # )
+
+ async def wait_for_application(self, model_name, application_name,
+ timeout=300):
"""Wait for an application to become active."""
if not self.authenticated:
await self.login()
- app = await self.get_application(self.default_model, name)
+ # TODO: In a point release, we will use a model per deployed network
+ # service. In the meantime, we will always use the 'default' model.
+ model_name = 'default'
+ model = await self.get_model(model_name)
+
+ app = await self.get_application(model, application_name)
+ self.log.debug("Application: {}".format(app))
+ # app = await self.get_application(model_name, application_name)
if app:
self.log.debug(
"JujuApi: Waiting {} seconds for Application {}".format(
timeout,
- name,
+ application_name,
)
)
- await self.default_model.block_until(
+ await model.block_until(
lambda: all(
- unit.agent_status == 'idle'
- and unit.workload_status
- in ['active', 'unknown'] for unit in app.units
+ unit.agent_status == 'idle' and unit.workload_status in
+ ['active', 'unknown'] for unit in app.units
),
timeout=timeout
)
projects / osm / N2VC.git / blobdiff