rwlaunchpad/plugins/rwvns/rift/tasklets/rwvnstasklet/rwvnstasklet.py

   1
   2 #
   3 #   Copyright 2016-2017 RIFT.IO Inc
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17
  18 import asyncio
  19 import sys
  20
  21 import gi
  22 gi.require_version('RwVnsYang', '1.0')
  23 gi.require_version('RwDts', '1.0')
  24 from gi.repository import (
  25     RwVnsYang,
  26     RwDts as rwdts,
  27     RwTypes,
  28 )
  29
  30 import rift.tasklets
  31 from rift.mano.utils.project import (
  32     ManoProject,
  33     ProjectHandler,
  34 )
  35 import rift.mano.sdn
  36
  37 from rift.vlmgr import (
  38     VlrDtsHandler,
  39     VldDtsHandler,
  40     VirtualLinkRecord,
  41     VirtualLinkEventListener
  42 )
  43
  44 from rift.topmgr import (
  45     NwtopStaticDtsHandler,
  46     NwtopDiscoveryDtsHandler,
  47     NwtopDataStore,
  48 )
  49
  50
  51 class VlRecordError(Exception):
  52     """ Vlr Record creation Error """
  53     pass
  54
  55
  56 class VlRecordNotFound(Exception):
  57     """ Vlr Record not found"""
  58     pass
  59
  60
  61 class SDNAccountHandlers(object):
  62     def __init__(self, dts, log, log_hdl, acctstore, loop, project):
  63         self._log = log
  64         self._log_hdl = log_hdl
  65         self._dts = dts
  66         self._loop = loop
  67         self._acctstore = acctstore
  68         self._project = project
  69
  70         self._log.debug("Creating SDN account config handler")
  71         self.sdn_cfg_handler = rift.mano.sdn.SDNAccountConfigSubscriber(
  72               self._dts, self._log, project, self._log_hdl,
  73               rift.mano.sdn.SDNAccountConfigCallbacks(
  74                   on_add_apply=self.on_sdn_account_added,
  75                   on_delete_apply=self.on_sdn_account_deleted,
  76               ),
  77               self._acctstore
  78
  79         )
  80
  81         self._log.debug("Creating SDN account opdata handler")
  82         self.sdn_operdata_handler = rift.mano.sdn.SDNAccountDtsOperdataHandler(
  83               self._dts, self._log, self._loop, project,
  84         )
  85
  86     def on_sdn_account_deleted(self, account_name):
  87         self._log.debug("SDN account deleted")
  88         self.sdn_operdata_handler.delete_sdn_account(account_name)
  89
  90     def on_sdn_account_added(self, account):
  91         self._log.debug("SDN account added")
  92         self.sdn_operdata_handler.add_sdn_account(account)
  93
  94     @asyncio.coroutine
  95     def register(self):
  96         self.sdn_cfg_handler.register()
  97         yield from self.sdn_operdata_handler.register()
  98
  99     def deregister(self):
 100         self.sdn_cfg_handler.deregister()
 101         self.sdn_operdata_handler.deregister()
 102
 103
 104 class VnsManager(object):
 105     """ The Virtual Network Service Manager """
 106     def __init__(self, dts, log, log_hdl, loop, project):
 107         self._dts = dts
 108         self._log = log
 109         self._log_hdl = log_hdl
 110         self._loop = loop
 111         self._project = project
 112         self._acctstore = {}
 113         self._vlr_handler = VlrDtsHandler(dts, log, loop, self)
 114         self._vld_handler = VldDtsHandler(dts, log, loop, self)
 115         self._sdn_handlers = SDNAccountHandlers(dts, log, log_hdl, self._acctstore, loop, self._project)
 116         self._nwtopdata_store = NwtopDataStore(log)
 117         self._nwtopdiscovery_handler = NwtopDiscoveryDtsHandler(dts, log, loop, project,
 118                                                                 self._acctstore, self._nwtopdata_store)
 119         self._nwtopstatic_handler = NwtopStaticDtsHandler(dts, log, loop, project,
 120                                                           self._acctstore, self._nwtopdata_store)
 121         self._vl_event_listener = VirtualLinkEventListener(dts, log, loop, self)
 122         self._vlrs = {}
 123
 124     @asyncio.coroutine
 125     def register_vlr_handler(self):
 126         """ Register vlr DTS handler """
 127         self._log.debug("Registering  DTS VLR handler")
 128         yield from self._vlr_handler.register()
 129
 130     @asyncio.coroutine
 131     def register_vld_handler(self):
 132         """ Register vlr DTS handler """
 133         self._log.debug("Registering  DTS VLD handler")
 134         yield from self._vld_handler.register()
 135
 136     @asyncio.coroutine
 137     def register_sdn_handlers(self):
 138         """ Register SDN DTS handlers """
 139         self._log.debug("Registering  SDN Account handlers")
 140         yield from self._sdn_handlers.register()
 141
 142     @asyncio.coroutine
 143     def register_nwtopstatic_handler(self):
 144         """ Register static NW topology DTS handler """
 145         self._log.debug("Registering  static DTS NW topology handler")
 146         yield from self._nwtopstatic_handler.register()
 147
 148     @asyncio.coroutine
 149     def register_nwtopdiscovery_handler(self):
 150         """ Register discovery-based NW topology DTS handler """
 151         self._log.debug("Registering  discovery-based DTS NW topology handler")
 152         yield from self._nwtopdiscovery_handler.register()
 153
 154     @asyncio.coroutine
 155     def register_vl_event_listener(self):
 156         """ Register Virtual Link related events DTS handler """
 157         self._log.debug("Registering  Virtual Link Event listener")
 158         yield from self._vl_event_listener.register()
 159
 160     @asyncio.coroutine
 161     def register(self):
 162         """ Register all static DTS handlers"""
 163         yield from self.register_sdn_handlers()
 164         yield from self.register_vlr_handler()
 165         yield from self.register_vld_handler()
 166         yield from self.register_nwtopstatic_handler()
 167         yield from self.register_nwtopdiscovery_handler()
 168         yield from self.register_vl_event_listener()
 169
 170     def deregister(self):
 171         self._vl_event_listener.deregister()
 172         self._nwtopdiscovery_handler.deregister()
 173         self._nwtopstatic_handler.deregister()
 174         self._vld_handler.deregister()
 175         self._vlr_handler.deregister()
 176         self._sdn_handlers.deregister()
 177
 178     def create_vlr(self, msg):
 179         """ Create VLR """
 180         if msg.id in self._vlrs:
 181             err = "Vlr id %s already exists" % msg.id
 182             self._log.error(err)
 183             # raise VlRecordError(err)
 184             return self._vlrs[msg.id]
 185
 186         self._log.info("Creating VirtualLinkRecord %s", msg.id)
 187         self._vlrs[msg.id] = VirtualLinkRecord(self._dts,
 188                                                self._log,
 189                                                self._loop,
 190                                                self,
 191                                                msg,
 192                                                )
 193         return self._vlrs[msg.id]
 194
 195     def get_vlr(self, vlr_id):
 196         """  Get VLR by vlr id """
 197         return self._vlrs[vlr_id]
 198
 199     @asyncio.coroutine
 200     def delete_vlr(self, vlr_id, xact):
 201         """ Delete VLR with the passed id"""
 202         if vlr_id not in self._vlrs:
 203             err = "Delete Failed - Vlr id %s not found" % vlr_id
 204             self._log.error(err)
 205             raise VlRecordNotFound(err)
 206
 207         self._log.info("Deleting virtual link id %s", vlr_id)
 208         yield from self._vlrs[vlr_id].terminate(xact)
 209         del self._vlrs[vlr_id]
 210         self._log.info("Deleted virtual link id %s", vlr_id)
 211
 212     def find_vlR_by_vld_id(self, vld_id):
 213         """ Find a VLR matching the VLD Id """
 214         for vlr in self._vlrs.values():
 215             if vlr.vld_id == vld_id:
 216                 return vlr
 217         return None
 218
 219     @asyncio.coroutine
 220     def run(self):
 221         """ Run this VNSM instance """
 222         self._log.debug("Run VNSManager - registering static DTS handlers")
 223         yield from self.register()
 224
 225     def vld_in_use(self, vld_id):
 226         """ Is this VLD in use """
 227         return False
 228
 229     @asyncio.coroutine
 230     def publish_vlr(self, xact, xpath, msg):
 231         """ Publish a VLR """
 232         path = self._project.add_project(xpath)
 233         self._log.debug("Publish vlr called with path %s, msg %s",
 234                         path, msg)
 235         yield from self._vlr_handler.update(xact, path, msg)
 236
 237     @asyncio.coroutine
 238     def unpublish_vlr(self, xact, xpath):
 239         """ Publish a VLR """
 240         path = self._project.add_project(xpath)
 241         self._log.debug("Unpublish vlr called with path %s", path)
 242         yield from self._vlr_handler.delete(xact, path)
 243
 244     def create_virual_link_event(self, event_id, event_msg):
 245         """ Update Virtual Link Event """
 246         self._log.debug("Creating Virtual Link Event id [%s], msg [%s]",
 247                        event_id, event_msg)
 248
 249     @asyncio.coroutine
 250     def update_virual_link_event(self, event_id, event_msg):
 251         """ Update Virtual Link Event """
 252         self._log.debug("Updating Virtual Link Event id [%s], msg [%s]",
 253                         event_id, event_msg)
 254         # event id and vlr_id are the same.
 255         # Use event id to look up the VLR and update and publish state change
 256         vlr = None
 257
 258         if event_id in self._vlrs:
 259             vlr = self._vlrs[event_id]
 260
 261         if vlr is None:
 262             self._log.error("Received VLR Event notifcation for unknown VLR - event-id:%s",
 263                             event_id)
 264             return
 265
 266         if event_msg.resource_info.resource_state == 'active':
 267             with self._dts.transaction(flags=0) as xact:
 268                 yield from vlr.ready(event_msg, xact)
 269         elif event_msg.resource_info.resource_state == 'failed':
 270             with self._dts.transaction(flags=0) as xact:
 271                 if event_msg.resource_info.resource_errors:
 272                     vlr._state_failed_reason = str(event_msg.resource_info.resource_errors)
 273                 yield from vlr.failed(event_msg, xact)
 274         else:
 275             self._log.warning("Receieved unknown resource state %s for event id %s vlr:%s",
 276                               event_msg.resource_info.resource_state, event_id, vlr.name)
 277
 278     def delete_virual_link_event(self, event_id):
 279         """ Delete Virtual Link Event """
 280         self._log.debug("Deleting Virtual Link Event id [%s]",
 281                         event_id)
 282
 283
 284 class VnsProject(ManoProject):
 285
 286     def __init__(self, name, tasklet, **kw):
 287         super(VnsProject, self).__init__(tasklet.log, name)
 288         self.update(tasklet)
 289
 290         self._vlr_handler = None
 291         self._vnsm = None
 292         # A mapping of instantiated vlr_id's to VirtualLinkRecord objects
 293         self._vlrs = {}
 294
 295     @asyncio.coroutine
 296     def register (self):
 297         try:
 298             self._vnsm = VnsManager(dts=self._dts,
 299                                     log=self.log,
 300                                     log_hdl=self.log_hdl,
 301                                     loop=self.loop,
 302                                     project=self)
 303             yield from self._vnsm.run()
 304         except Exception as e:
 305             self.log.exception("VNS Task failed to run", e)
 306
 307     def deregister(self):
 308         self._log.debug("De-register project {}".format(self.name))
 309         self._vnsm.deregister()
 310
 311
 312 class VnsTasklet(rift.tasklets.Tasklet):
 313     """ The VNS tasklet class """
 314     def __init__(self, *args, **kwargs):
 315         super(VnsTasklet, self).__init__(*args, **kwargs)
 316         self.rwlog.set_category("rw-mano-log")
 317         self.rwlog.set_subcategory("vns")
 318
 319         self._dts = None
 320         self._project_handler = None
 321         self.projects = {}
 322
 323     @property
 324     def dts(self):
 325         return self._dts
 326
 327     def start(self):
 328         super(VnsTasklet, self).start()
 329         self.log.info("Starting VnsTasklet")
 330
 331         self.log.debug("Registering with dts")
 332         try:
 333             self._dts = rift.tasklets.DTS(self.tasklet_info,
 334                                           RwVnsYang.get_schema(),
 335                                           self.loop,
 336                                           self.on_dts_state_change)
 337         except Exception:
 338             self.log.exception("Caught Exception in VNS start:", e)
 339
 340         self.log.debug("Created DTS Api GI Object: %s", self._dts)
 341
 342     def on_instance_started(self):
 343         """ The task instance started callback"""
 344         self.log.debug("Got instance started callback")
 345
 346     def stop(self):
 347       try:
 348          self._dts.deinit()
 349       except Exception:
 350          print("Caught Exception in VNS stop:", sys.exc_info()[0])
 351          raise
 352
 353     @asyncio.coroutine
 354     def init(self):
 355         """ task init callback"""
 356         self.log.debug("creating project handler")
 357         self.project_handler = ProjectHandler(self, VnsProject)
 358         self.project_handler.register()
 359
 360     @asyncio.coroutine
 361     def run(self):
 362         """ tasklet run callback """
 363         pass
 364
 365     @asyncio.coroutine
 366     def on_dts_state_change(self, state):
 367         """Take action according to current dts state to transition
 368         application into the corresponding application state
 369
 370         Arguments
 371             state - current dts state
 372         """
 373         switch = {
 374             rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
 375             rwdts.State.CONFIG: rwdts.State.RUN,
 376         }
 377
 378         handlers = {
 379             rwdts.State.INIT: self.init,
 380             rwdts.State.RUN: self.run,
 381         }
 382
 383         # Transition application to next state
 384         handler = handlers.get(state, None)
 385         if handler is not None:
 386             yield from handler()
 387
 388         # Transition dts to next state
 389         next_state = switch.get(state, None)
 390         if next_state is not None:
 391             self._dts.handle.set_state(next_state)