Improve event monitoring/callbacks

[osm/N2VC.git] / n2vc / vnf.py

diff --git a/n2vc/vnf.py b/n2vc/vnf.py
index c606dda..a457cec 100644 (file)
--- a/n2vc/vnf.py
+++ b/n2vc/vnf.py
@@ -36,6 +36,9 @@ class JujuCharmNotFound(Exception):
 class JujuApplicationExists(Exception):
     """The Application already exists."""
 
 class JujuApplicationExists(Exception):
     """The Application already exists."""
 

+class N2VCPrimitiveExecutionFailed(Exception):
+    """Something failed while attempting to execute a primitive."""
+

 
 # Quiet the debug logging
 logging.getLogger('websockets.protocol').setLevel(logging.INFO)
 
 # Quiet the debug logging
 logging.getLogger('websockets.protocol').setLevel(logging.INFO)


@@ -45,41 +48,83 @@ logging.getLogger('juju.machine').setLevel(logging.WARN)
 
 class VCAMonitor(ModelObserver):
     """Monitor state changes within the Juju Model."""
 
 class VCAMonitor(ModelObserver):
     """Monitor state changes within the Juju Model."""

-    callback = None
-    callback_args = None

     log = None
     ns_name = None
     log = None
     ns_name = None

-    application_name = None
+    applications = {}

 
 

-    def __init__(self, ns_name, application_name, callback, *args):
+    def __init__(self, ns_name):

         self.log = logging.getLogger(__name__)
 
         self.ns_name = ns_name
         self.log = logging.getLogger(__name__)
 
         self.ns_name = ns_name

-        self.application_name = application_name
-        self.callback = callback
-        self.callback_args = args
+
+    def AddApplication(self, application_name, callback, *callback_args):
+        if application_name not in self.applications:
+            self.applications[application_name] = {
+                'callback': callback,
+                'callback_args': callback_args
+            }
+
+    def RemoveApplication(self, application_name):
+        if application_name in self.applications:
+            del self.applications[application_name]

 
     async def on_change(self, delta, old, new, model):
         """React to changes in the Juju model."""
 
         if delta.entity == "unit":
 
     async def on_change(self, delta, old, new, model):
         """React to changes in the Juju model."""
 
         if delta.entity == "unit":

+            # Ignore change events from other applications
+            if delta.data['application'] not in self.applications.keys():
+                return
+

             try:
             try:

+
+                application_name = delta.data['application']
+
+                callback = self.applications[application_name]['callback']
+                callback_args = self.applications[application_name]['callback_args']
+

                 if old and new:
                     old_status = old.workload_status
                     new_status = new.workload_status
                 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 old_status == new_status:
                         """The workload status may fluctuate around certain events,
                         so wait until the status has stabilized before triggering
                         the callback."""

-                        if self.callback:
-                            self.callback(
+                        if callback:
+                            callback(

                                 self.ns_name,
                                 self.ns_name,

-                                self.application_name,
+                                delta.data['application'],

                                 new_status,
                                 new_status,

-                                *self.callback_args)
+                                *callback_args)
+
+                if old and not new:
+                    # This is a charm being removed
+                    if callback:
+                        callback(
+                            self.ns_name,
+                            delta.data['application'],
+                            "removed",
+                            *callback_args)

             except Exception as e:
                 self.log.debug("[1] notify_callback exception {}".format(e))
             except Exception as 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
+
+            # uuid = delta.data['id']     # The Action's unique id
+            # msg = delta.data['message'] # The output of the action
+            #
+            # if delta.data['status'] == "pending":
+            #     # The action is queued
+            #     pass
+            # elif delta.data['status'] == "completed""
+            #     # The action was successful
+            #     pass
+            # elif delta.data['status'] == "failed":
+            #     # The action failed.
+            #     pass

 
 

+            pass

 
 ########
 # TODO
 
 ########
 # TODO


@@ -170,6 +215,7 @@ class N2VC:
                 callback(model_name, application_name, status, *callback_args)
         except Exception as e:
             self.log.error("[0] notify_callback exception {}".format(e))
                 callback(model_name, application_name, status, *callback_args)
         except Exception as e:
             self.log.error("[0] notify_callback exception {}".format(e))

+            raise e

         return True
 
     # Public methods
         return True
 
     # Public methods


@@ -207,7 +253,9 @@ class N2VC:
         :param dict params: A dictionary of runtime parameters
           Examples::
           {
         :param dict params: A dictionary of runtime parameters
           Examples::
           {

-            'rw_mgmt_ip': '1.2.3.4'
+            'rw_mgmt_ip': '1.2.3.4',
+            # 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
           Examples::
           }
         :param dict machine_spec: A dictionary describing the machine to install to
           Examples::


@@ -241,18 +289,6 @@ class N2VC:
         # service. In the meantime, we will always use the 'default' model.
         model_name = 'default'
         model = await self.get_model(model_name)
         # service. In the meantime, we will always use the 'default' model.
         model_name = 'default'
         model = await self.get_model(model_name)

-        # if model_name not in self.models:
-        #     self.log.debug("Getting model {}".format(model_name))
-        #     self.models[model_name] = await self.controller.get_model(model_name)
-        # model = await self.CreateNetworkService(ns_name)
-
-        ###################################################
-        # Get the name of the charm and its configuration #
-        ###################################################
-        config_dict = vnfd['vnf-configuration']
-        juju = config_dict['juju']
-        charm = juju['charm']
-        self.log.debug("Charm: {}".format(charm))

 
         ########################################
         # Verify the application doesn't exist #
 
         ########################################
         # Verify the application doesn't exist #


@@ -261,14 +297,15 @@ class N2VC:
         if app:
             raise JujuApplicationExists("Can't deploy application \"{}\" to model \"{}\" because it already exists.".format(application_name, model))
 
         if app:
             raise JujuApplicationExists("Can't deploy application \"{}\" to model \"{}\" because it already exists.".format(application_name, model))
 

-        ############################################################
-        # Create a monitor to watch for application status changes #
-        ############################################################
+        ################################################################
+        # Register this application with the model-level event monitor #
+        ################################################################

         if callback:
         if callback:

-            self.log.debug("Setting monitor<->callback")
-            self.monitors[application_name] = VCAMonitor(model_name, application_name, callback, *callback_args)
-            model.add_observer(self.monitors[application_name])
-
+            self.monitors[model_name].AddApplication(
+                application_name,
+                callback,
+                *callback_args
+            )

 
         ########################################################
         # Check for specific machine placement (native charms) #
 
         ########################################################
         # Check for specific machine placement (native charms) #


@@ -294,12 +331,11 @@ class N2VC:
             rw_mgmt_ip = params['rw_mgmt_ip']
 
         initial_config = self._get_config_from_dict(
             rw_mgmt_ip = params['rw_mgmt_ip']
 
         initial_config = self._get_config_from_dict(

-            config_dict['initial-config-primitive'],
+            params['initial-config-primitive'],

             {'<rw_mgmt_ip>': rw_mgmt_ip}
         )
 
             {'<rw_mgmt_ip>': rw_mgmt_ip}
         )
 

-        self.log.debug("JujuApi: Deploying charm {} ({}) from {}".format(
-            charm,
+        self.log.debug("JujuApi: Deploying charm ({}) from {}".format(

             application_name,
             charm_path,
             to=to,
             application_name,
             charm_path,
             to=to,


@@ -309,13 +345,57 @@ class N2VC:
         # Deploy the charm and apply the initial configuration #
         ########################################################
         app = await model.deploy(
         # Deploy the charm and apply the initial configuration #
         ########################################################
         app = await model.deploy(

+            # We expect charm_path to be either the path to the charm on disk
+            # or in the format of cs:series/name

             charm_path,
             charm_path,

+            # This is the formatted, unique name for this charm

             application_name=application_name,
             application_name=application_name,

+            # Proxy charms should use the current LTS. This will need to be
+            # changed for native charms.

             series='xenial',
             series='xenial',

+            # Apply the initial 'config' primitive during deployment

             config=initial_config,
             config=initial_config,

+            # TBD: Where to deploy the charm to.

             to=None,
         )
 
             to=None,
         )
 

+        # #######################################
+        # # Execute initial config primitive(s) #
+        # #######################################
+        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']
+
+                    primitives[seq] = {
+                        'name': primitive['name'],
+                        'parameters': self._map_primitive_parameters(
+                            primitive['parameter'],
+                            {'<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.debug.log(
+                    "[N2VC] Exception executing primitive: {}".format(e)
+                )
+                raise
+

     async def ExecutePrimitive(self, model_name, application_name, primitive, callback, *callback_args, **params):
         try:
             if not self.authenticated:
     async def ExecutePrimitive(self, model_name, application_name, primitive, callback, *callback_args, **params):
         try:
             if not self.authenticated:


@@ -336,7 +416,7 @@ class N2VC:
                     if unit:
                         self.log.debug("Executing primitive {}".format(primitive))
                         action = await unit.run_action(primitive, **params)
                     if unit:
                         self.log.debug("Executing primitive {}".format(primitive))
                         action = await unit.run_action(primitive, **params)

-                        action = await action.wait()
+                        # action = await action.wait()

                 await model.disconnect()
         except Exception as e:
             self.log.debug("Caught exception while executing primitive: {}".format(e))
                 await model.disconnect()
         except Exception as e:
             self.log.debug("Caught exception while executing primitive: {}".format(e))


@@ -350,11 +430,19 @@ class N2VC:
             model = await self.get_model(model_name)
             app = await self.get_application(model, application_name)
             if app:
             model = await self.get_model(model_name)
             app = await self.get_application(model, application_name)
             if app:

-                self.notify_callback(model_name, application_name, "removing", callback, *callback_args)
+                # Remove this application from event monitoring
+                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))

                 await app.remove()
                 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))
         except Exception as e:
             print("Caught exception: {}".format(e))

+            self.log.debug(e)

             raise e
 
     async def DestroyNetworkService(self, nsd):
             raise e
 
     async def DestroyNetworkService(self, nsd):


@@ -391,10 +479,18 @@ class N2VC:
         return await self.set_config(application=application, config=config)
 
     def _get_config_from_dict(self, config_primitive, values):
         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."""
+        """Transform the yang config primitive to dict.
+
+        Expected result:
+
+            config = {
+                'config':
+            }
+        """

         config = {}
         for primitive in config_primitive:
             if primitive['name'] == 'config':
         config = {}
         for primitive in config_primitive:
             if primitive['name'] == 'config':

+                # config = self._map_primitive_parameters()

                 for parameter in primitive['parameter']:
                     param = str(parameter['name'])
                     if parameter['value'] == "<rw_mgmt_ip>":
                 for parameter in primitive['parameter']:
                     param = str(parameter['name'])
                     if parameter['value'] == "<rw_mgmt_ip>":


@@ -404,6 +500,16 @@ class N2VC:
 
         return config
 
 
         return config
 

+    def _map_primitive_parameters(self, parameters, values):
+        params = {}
+        for parameter in parameters:
+            param = str(parameter['name'])
+            if parameter['value'] == "<rw_mgmt_ip>":
+                params[param] = str(values[parameter['value']])
+            else:
+                params[param] = str(parameter['value'])
+        return params
+

     def _get_config_from_yang(self, config_primitive, values):
         """Transform the yang config primitive to dict."""
         config = {}
     def _get_config_from_yang(self, config_primitive, values):
         """Transform the yang config primitive to dict."""
         config = {}


@@ -494,6 +600,10 @@ class N2VC:
             print("connecting to model {}".format(model_name))
             self.models[model_name] = await self.controller.get_model(model_name)
 
             print("connecting to model {}".format(model_name))
             self.models[model_name] = await self.controller.get_model(model_name)
 

+            # Create an observer for this model
+            self.monitors[model_name] = VCAMonitor(model_name)
+            self.models[model_name].add_observer(self.monitors[model_name])
+

         return self.models[model_name]
 
     async def login(self):
         return self.models[model_name]
 
     async def login(self):


@@ -521,6 +631,11 @@ class N2VC:
             # current_controller no longer exists
             # self.log.debug("Connecting to current controller...")
             # await self.controller.connect_current()
             # current_controller no longer exists
             # self.log.debug("Connecting to current controller...")
             # await self.controller.connect_current()

+            # await self.controller.connect(
+            #     endpoint=self.endpoint,
+            #     username=self.user,
+            #     cacert=cacert,
+            # )

             self.log.fatal("VCA credentials not configured.")
 
         self.authenticated = True
             self.log.fatal("VCA credentials not configured.")
 
         self.authenticated = True