rwlaunchpad/plugins/rwvns/rift/topmgr/rwtopmgr.py

   1
   2 #
   3 #   Copyright 2016 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
  20 import gi
  21 gi.require_version('RwDts', '1.0')
  22 gi.require_version('RwcalYang', '1.0')
  23 gi.require_version('RwTypes', '1.0')
  24 gi.require_version('RwSdn', '1.0')
  25 from gi.repository import (
  26     RwDts as rwdts,
  27     IetfNetworkYang,
  28     IetfNetworkTopologyYang,
  29     IetfL2TopologyYang,
  30     RwTopologyYang,
  31     RwsdnalYang,
  32     RwTypes
  33 )
  34
  35 from gi.repository.RwTypes import RwStatus
  36 import rift.tasklets
  37
  38
  39 class NwtopDiscoveryDtsHandler(object):
  40     """ Handles DTS interactions for the Discovered Topology registration """
  41     DISC_XPATH = "D,/nd:network"
  42
  43     def __init__(self, dts, log, loop, project, acctstore, nwdatastore):
  44         self._dts = dts
  45         self._log = log
  46         self._loop = loop
  47         self._project = project
  48         self._acctstore = acctstore
  49         self._nwdatastore = nwdatastore
  50
  51         self._regh = None
  52
  53     @property
  54     def regh(self):
  55         """ The registration handle associated with this Handler"""
  56         return self._regh
  57
  58     def deregister(self):
  59         self._log.debug("De-register Topology discovery handler for project {}".
  60                         format(self._project.name))
  61         if self._regh:
  62             self._regh.deregister()
  63             self._regh = None
  64
  65     @asyncio.coroutine
  66     def register(self):
  67         """ Register for the Discovered Topology path """
  68
  69         @asyncio.coroutine
  70         def on_ready(regh, status):
  71             """  On_ready for Discovered Topology registration """
  72             self._log.debug("PUB reg ready for Discovered Topology handler regn_hdl(%s) status %s",
  73                                          regh, status)
  74
  75         @asyncio.coroutine
  76         def on_prepare(xact_info, action, ks_path, msg):
  77             """ prepare for Discovered Topology registration"""
  78             self._log.debug(
  79                 "Got topology on_prepare callback (xact_info: %s, action: %s): %s",
  80                 xact_info, action, msg
  81                 )
  82
  83             if action == rwdts.QueryAction.READ:
  84
  85                 for name, sdnacct in self._acctstore.items():
  86                     if sdnacct.account_type != "odl":
  87                         continue
  88                     sdnintf = sdnacct.sdn
  89
  90                     rc, nwtop = sdnintf.get_network_list(sdnacct.sdnal_account_msg)
  91                     #assert rc == RwStatus.SUCCESS
  92                     if rc != RwStatus.SUCCESS:
  93                         self._log.error("Fetching get network list for SDN Account %s failed", name)
  94                         xact_info.respond_xpath(rwdts.XactRspCode.NACK)
  95                         return
  96
  97                     self._log.debug("Topology: Retrieved network attributes ")
  98                     for nw in nwtop.network:
  99                         # Add SDN account name
 100                         nw.rw_network_attributes.sdn_account_name = name
 101                         nw.server_provided = False
 102                         nw.network_id = name + ':' + nw.network_id
 103                         self._log.debug("...Network id %s", nw.network_id)
 104                         nw_xpath = ("D,/nd:network[network-id=\'{}\']").format(nw.network_id)
 105                         xact_info.respond_xpath(rwdts.XactRspCode.MORE,
 106                                         nw_xpath, nw)
 107
 108                 xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 109                 #err = "%s action on discovered Topology not supported" % action
 110                 #raise NotImplementedError(err)
 111
 112         self._log.debug("Registering for discovered topology using xpath %s", NwtopDiscoveryDtsHandler.DISC_XPATH)
 113
 114         handler = rift.tasklets.DTS.RegistrationHandler(
 115             on_ready=on_ready,
 116             on_prepare=on_prepare,
 117             )
 118
 119         yield from self._dts.register(
 120             NwtopDiscoveryDtsHandler.DISC_XPATH,
 121             flags=rwdts.Flag.PUBLISHER,
 122             handler=handler
 123             )
 124
 125
 126 class NwtopStaticDtsHandler(object):
 127     """ Handles DTS interactions for the Static Topology registration """
 128     STATIC_XPATH = "C,/nd:network"
 129
 130     def __init__(self, dts, log, loop, project, acctstore, nwdatastore):
 131         self._dts = dts
 132         self._log = log
 133         self._loop = loop
 134         self._project = project
 135         self._acctstore = acctstore
 136
 137         self._regh = None
 138         self.pending = {}
 139         self._nwdatastore = nwdatastore
 140
 141     @property
 142     def regh(self):
 143         """ The registration handle associated with this Handler"""
 144         return self._regh
 145
 146     def deregister(self):
 147         self._log.debug("De-register Topology static handler for project {}".
 148                         format(self._project.name))
 149         if self._regh:
 150             self._regh.deregister()
 151             self._regh = None
 152
 153     @asyncio.coroutine
 154     def register(self):
 155         """ Register for the Static Topology path """
 156
 157         @asyncio.coroutine
 158         def prepare_nw_cfg(dts, acg, xact, xact_info, ksp, msg, scratch):
 159             """Prepare for application configuration. Stash the pending
 160             configuration object for subsequent transaction phases"""
 161             self._log.debug("Prepare Network config received network id %s, msg %s",
 162                            msg.network_id, msg)
 163             self.pending[xact.id] = msg
 164             xact_info.respond_xpath(rwdts.XactRspCode.ACK)
 165
 166         def apply_nw_config(dts, acg, xact, action, scratch):
 167             """Apply the pending configuration object"""
 168             if action == rwdts.AppconfAction.INSTALL and xact.id is None:
 169                 self._log.debug("No xact handle.  Skipping apply config")
 170                 return
 171
 172             if xact.id not in self.pending:
 173                 raise KeyError("No stashed configuration found with transaction id [{}]".format(xact.id))
 174
 175             try:
 176                 if action == rwdts.AppconfAction.INSTALL:
 177                     self._nwdatastore.create_network(self.pending[xact.id].network_id, self.pending[xact.id])
 178                 elif action == rwdts.AppconfAction.RECONCILE:
 179                     self._nwdatastore.update_network(self.pending[xact.id].network_id, self.pending[xact.id])
 180             except:
 181                 raise
 182
 183             self._log.debug("Create network config done")
 184             return RwTypes.RwStatus.SUCCESS
 185
 186         self._log.debug("Registering for static topology using xpath %s", NwtopStaticDtsHandler.STATIC_XPATH)
 187         handler=rift.tasklets.AppConfGroup.Handler(
 188                         on_apply=apply_nw_config)
 189
 190         with self._dts.appconf_group_create(handler=handler) as acg:
 191             self._regh = acg.register(xpath = NwtopStaticDtsHandler.STATIC_XPATH,
 192                                       flags = rwdts.Flag.SUBSCRIBER,
 193                                       on_prepare=prepare_nw_cfg)