| #!/usr/bin/env python3 |
| |
| # |
| # Copyright 2016 RIFT.IO Inc |
| # |
| # 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 argparse |
| import asyncio |
| import logging |
| import os |
| import sys |
| import time |
| import unittest |
| import uuid |
| |
| import xmlrunner |
| |
| import gi.repository.RwDts as rwdts |
| import gi.repository.RwNsmYang as rwnsmyang |
| import gi.repository.NsrYang as NsrYang |
| import gi.repository.RwNsrYang as RwNsrYang |
| import gi.repository.RwTypes as RwTypes |
| import gi.repository.ProtobufC as ProtobufC |
| import gi.repository.RwResourceMgrYang as RwResourceMgrYang |
| import gi.repository.RwLaunchpadYang as launchpadyang |
| import rift.tasklets |
| import rift.test.dts |
| |
| import mano_ut |
| |
| |
| if sys.version_info < (3, 4, 4): |
| asyncio.ensure_future = asyncio.async |
| |
| |
| class NsrDtsHandler(object): |
| """ The network service DTS handler """ |
| NSR_XPATH = "C,/nsr:ns-instance-config/nsr:nsr" |
| SCALE_INSTANCE_XPATH = "C,/nsr:ns-instance-config/nsr:nsr/nsr:scaling-group/nsr:instance" |
| |
| def __init__(self, dts, log, loop, nsm): |
| self._dts = dts |
| self._log = log |
| self._loop = loop |
| self._nsm = nsm |
| |
| self._nsr_regh = None |
| self._scale_regh = None |
| |
| @property |
| def nsm(self): |
| """ Return the NS manager instance """ |
| return self._nsm |
| |
| def get_scale_group_instances(self, nsr_id, group_name): |
| def nsr_id_from_keyspec(ks): |
| nsr_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks) |
| nsr_id = nsr_path_entry.key00.id |
| return nsr_id |
| |
| def group_name_from_keyspec(ks): |
| group_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks) |
| group_name = group_path_entry.key00.scaling_group_name_ref |
| return group_name |
| |
| |
| xact_ids = set() |
| for instance_cfg, keyspec in self._scale_regh.get_xact_elements(include_keyspec=True): |
| elem_nsr_id = nsr_id_from_keyspec(keyspec) |
| if elem_nsr_id != nsr_id: |
| continue |
| |
| elem_group_name = group_name_from_keyspec(keyspec) |
| if elem_group_name != group_name: |
| continue |
| |
| xact_ids.add(instance_cfg.id) |
| |
| return xact_ids |
| |
| @asyncio.coroutine |
| def register(self): |
| """ Register for Nsr create/update/delete/read requests from dts """ |
| |
| def nsr_id_from_keyspec(ks): |
| nsr_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr.schema().keyspec_to_entry(ks) |
| nsr_id = nsr_path_entry.key00.id |
| return nsr_id |
| |
| def group_name_from_keyspec(ks): |
| group_path_entry = NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup.schema().keyspec_to_entry(ks) |
| group_name = group_path_entry.key00.scaling_group_name_ref |
| return group_name |
| |
| def is_instance_in_reg_elements(nsr_id, group_name, instance_id): |
| """ Return boolean indicating if scaling group instance was already commited previously. |
| |
| By looking at the existing elements in this registration handle (elements not part |
| of this current xact), we can tell if the instance was configured previously without |
| keeping any application state. |
| """ |
| for instance_cfg, keyspec in self._nsr_regh.get_xact_elements(include_keyspec=True): |
| elem_nsr_id = nsr_id_from_keyspec(keyspec) |
| elem_group_name = group_name_from_keyspec(keyspec) |
| |
| if elem_nsr_id != nsr_id or group_name != elem_group_name: |
| continue |
| |
| if instance_cfg.id == instance_id: |
| return True |
| |
| return False |
| |
| def get_scale_group_instance_delta(nsr_id, group_name, xact): |
| |
| #1. Find all elements in the transaction add to the "added" |
| #2. Find matching elements in current elements, remove from "added". |
| #3. Find elements only in current, add to "deleted" |
| |
| xact_ids = set() |
| for instance_cfg, keyspec in self._scale_regh.get_xact_elements(xact, include_keyspec=True): |
| elem_nsr_id = nsr_id_from_keyspec(keyspec) |
| if elem_nsr_id != nsr_id: |
| continue |
| |
| elem_group_name = group_name_from_keyspec(keyspec) |
| if elem_group_name != group_name: |
| continue |
| |
| xact_ids.add(instance_cfg.id) |
| |
| current_ids = set() |
| for instance_cfg, keyspec in self._scale_regh.get_xact_elements(include_keyspec=True): |
| elem_nsr_id = nsr_id_from_keyspec(keyspec) |
| if elem_nsr_id != nsr_id: |
| continue |
| |
| elem_group_name = group_name_from_keyspec(keyspec) |
| if elem_group_name != group_name: |
| continue |
| |
| current_ids.add(instance_cfg.id) |
| |
| delta = { |
| "added": xact_ids - current_ids, |
| "deleted": current_ids - xact_ids |
| } |
| return delta |
| |
| def get_add_delete_update_cfgs(dts_member_reg, xact, key_name): |
| # Unforunately, it is currently difficult to figure out what has exactly |
| # changed in this xact without Pbdelta support (RIFT-4916) |
| # As a workaround, we can fetch the pre and post xact elements and |
| # perform a comparison to figure out adds/deletes/updates |
| xact_cfgs = list(dts_member_reg.get_xact_elements(xact)) |
| curr_cfgs = list(dts_member_reg.elements) |
| |
| xact_key_map = {getattr(cfg, key_name): cfg for cfg in xact_cfgs} |
| curr_key_map = {getattr(cfg, key_name): cfg for cfg in curr_cfgs} |
| |
| # Find Adds |
| added_keys = set(xact_key_map) - set(curr_key_map) |
| added_cfgs = [xact_key_map[key] for key in added_keys] |
| |
| # Find Deletes |
| deleted_keys = set(curr_key_map) - set(xact_key_map) |
| deleted_cfgs = [curr_key_map[key] for key in deleted_keys] |
| |
| # Find Updates |
| updated_keys = set(curr_key_map) & set(xact_key_map) |
| updated_cfgs = [xact_key_map[key] for key in updated_keys if xact_key_map[key] != curr_key_map[key]] |
| |
| return added_cfgs, deleted_cfgs, updated_cfgs |
| |
| def on_apply(dts, acg, xact, action, scratch): |
| """Apply the configuration""" |
| def handle_create_nsr(msg): |
| # Handle create nsr requests """ |
| # Do some validations |
| if not msg.has_field("nsd_ref"): |
| err = "NSD reference not provided" |
| self._log.error(err) |
| raise NetworkServiceRecordError(err) |
| |
| self._log.info("Creating NetworkServiceRecord %s from nsd_id %s", |
| msg.id, msg.nsd_ref) |
| |
| #nsr = self.nsm.create_nsr(msg) |
| return nsr |
| |
| def handle_delete_nsr(msg): |
| @asyncio.coroutine |
| def delete_instantiation(ns_id): |
| """ Delete instantiation """ |
| pass |
| #with self._dts.transaction() as xact: |
| #yield from self._nsm.terminate_ns(ns_id, xact) |
| |
| # Handle delete NSR requests |
| self._log.info("Delete req for NSR Id: %s received", msg.id) |
| # Terminate the NSR instance |
| #nsr = self._nsm.get_ns_by_nsr_id(msg.id) |
| |
| #nsr.set_state(NetworkServiceRecordState.TERMINATE_RCVD) |
| #event_descr = "Terminate rcvd for NS Id:%s" % msg.id |
| #nsr.record_event("terminate-rcvd", event_descr) |
| |
| #self._loop.create_task(delete_instantiation(msg.id)) |
| |
| @asyncio.coroutine |
| def begin_instantiation(nsr): |
| # Begin instantiation |
| pass |
| #self._log.info("Beginning NS instantiation: %s", nsr.id) |
| #yield from self._nsm.instantiate_ns(nsr.id, xact) |
| |
| self._log.debug("Got nsr apply (xact: %s) (action: %s)(scr: %s)", |
| xact, action, scratch) |
| |
| if action == rwdts.AppconfAction.INSTALL and xact.id is None: |
| self._log.debug("No xact handle. Skipping apply config") |
| xact = None |
| |
| (added_msgs, deleted_msgs, updated_msgs) = get_add_delete_update_cfgs(self._nsr_regh, xact, "id") |
| |
| for msg in added_msgs: |
| self._log.info("Create NSR received in on_apply to instantiate NS:%s", msg.id) |
| #if msg.id not in self._nsm.nsrs: |
| # self._log.info("Create NSR received in on_apply to instantiate NS:%s", msg.id) |
| # nsr = handle_create_nsr(msg) |
| # self._loop.create_task(begin_instantiation(nsr)) |
| |
| for msg in deleted_msgs: |
| self._log.info("Delete NSR received in on_apply to terminate NS:%s", msg.id) |
| try: |
| handle_delete_nsr(msg) |
| except Exception: |
| self._log.exception("Failed to terminate NS:%s", msg.id) |
| |
| for msg in updated_msgs: |
| self._log.info("Update NSR received in on_apply to change scaling groups in NS:%s", msg.id) |
| |
| for group in msg.scaling_group: |
| instance_delta = get_scale_group_instance_delta(msg.id, group.scaling_group_name_ref, xact) |
| self._log.debug("Got NSR:%s scale group instance delta: %s", msg.id, instance_delta) |
| |
| #for instance_id in instance_delta["added"]: |
| # self._nsm.scale_nsr_out(msg.id, group.scaling_group_name_ref, instance_id, xact) |
| |
| #for instance_id in instance_delta["deleted"]: |
| # self._nsm.scale_nsr_in(msg.id, group.scaling_group_name_ref, instance_id) |
| |
| |
| return RwTypes.RwStatus.SUCCESS |
| |
| @asyncio.coroutine |
| def on_prepare(dts, acg, xact, xact_info, ks_path, msg, scratch): |
| """ Prepare calllback from DTS for NSR """ |
| |
| xpath = ks_path.to_xpath(NsrYang.get_schema()) |
| action = xact_info.query_action |
| self._log.debug( |
| "Got Nsr prepare callback (xact: %s) (action: %s) (info: %s), %s:%s)", |
| xact, action, xact_info, xpath, msg |
| ) |
| |
| fref = ProtobufC.FieldReference.alloc() |
| fref.goto_whole_message(msg.to_pbcm()) |
| |
| if action in [rwdts.QueryAction.CREATE, rwdts.QueryAction.UPDATE]: |
| pass |
| # Ensure the Cloud account has been specified if this is an NSR create |
| #if msg.id not in self._nsm.nsrs: |
| # if not msg.has_field("cloud_account"): |
| # raise NsrInstantiationFailed("Cloud account not specified in NSR") |
| |
| # We do not allow scaling actions to occur if the NS is not in running state |
| #elif msg.has_field("scaling_group"): |
| # nsr = self._nsm.nsrs[msg.id] |
| # if nsr.state != NetworkServiceRecordState.RUNNING: |
| # raise ScalingOperationError("Unable to perform scaling action when NS is not in running state") |
| |
| # if len(msg.scaling_group) > 1: |
| # raise ScalingOperationError("Only a single scaling group can be configured at a time") |
| |
| # for group_msg in msg.scaling_group: |
| # num_new_group_instances = len(group_msg.instance) |
| # if num_new_group_instances > 1: |
| # raise ScalingOperationError("Only a single scaling instance can be created at a time") |
| |
| # elif num_new_group_instances == 1: |
| # scale_group = nsr.scaling_groups[group_msg.scaling_group_name_ref] |
| # if len(scale_group.instances) == scale_group.max_instance_count: |
| # raise ScalingOperationError("Max instances for %s reached" % scale_group) |
| |
| |
| acg.handle.prepare_complete_ok(xact_info.handle) |
| |
| |
| self._log.debug("Registering for NSR config using xpath: %s", |
| NsrDtsHandler.NSR_XPATH) |
| |
| acg_hdl = rift.tasklets.AppConfGroup.Handler(on_apply=on_apply) |
| with self._dts.appconf_group_create(handler=acg_hdl) as acg: |
| self._nsr_regh = acg.register(xpath=NsrDtsHandler.NSR_XPATH, |
| flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE, |
| on_prepare=on_prepare) |
| |
| self._scale_regh = acg.register( |
| xpath=NsrDtsHandler.SCALE_INSTANCE_XPATH, |
| flags=rwdts.Flag.SUBSCRIBER | rwdts.Flag.DELTA_READY | rwdts.Flag.CACHE, |
| ) |
| |
| |
| class XPaths(object): |
| @staticmethod |
| def nsr_config(nsr_id=None): |
| return ("C,/nsr:ns-instance-config/nsr:nsr" + |
| ("[nsr:id='{}']".format(nsr_id) if nsr_id is not None else "")) |
| |
| def scaling_group_instance(nsr_id, group_name, instance_id): |
| return ("C,/nsr:ns-instance-config/nsr:nsr" + |
| "[nsr:id='{}']".format(nsr_id) + |
| "/nsr:scaling-group" + |
| "[nsr:scaling-group-name-ref='{}']".format(group_name) + |
| "/nsr:instance" + |
| "[nsr:id='{}']".format(instance_id) |
| ) |
| |
| |
| class NsrHandlerTestCase(rift.test.dts.AbstractDTSTest): |
| """ |
| DTS GI interface unittests |
| """ |
| @classmethod |
| def configure_schema(cls): |
| return NsrYang.get_schema() |
| |
| @classmethod |
| def configure_timeout(cls): |
| return 240 |
| |
| def configure_test(self, loop, test_id): |
| self.log.debug("STARTING - %s", self.id()) |
| self.tinfo = self.new_tinfo(self.id()) |
| self.dts = rift.tasklets.DTS(self.tinfo, self.schema, self.loop) |
| self.handler = NsrDtsHandler(self.dts, self.log, self.loop, None) |
| |
| self.tinfo_c = self.new_tinfo(self.id() + "_client") |
| self.dts_c = rift.tasklets.DTS(self.tinfo_c, self.schema, self.loop) |
| |
| @rift.test.dts.async_test |
| def test_add_delete_ns(self): |
| |
| nsr1_uuid = "nsr1_uuid" # str(uuid.uuid4()) |
| nsr2_uuid = "nsr2_uuid" # str(uuid.uuid4()) |
| |
| assert nsr1_uuid != nsr2_uuid |
| |
| yield from self.handler.register() |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| self.log.debug("Creating NSR") |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_update( |
| XPaths.nsr_config(nsr1_uuid), |
| NsrYang.YangData_Nsr_NsInstanceConfig_Nsr(id=nsr1_uuid, name="fu"), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_update( |
| XPaths.scaling_group_instance(nsr1_uuid, "group", 1234), |
| NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance(id=1234), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_delete( |
| XPaths.scaling_group_instance(nsr1_uuid, "group", 1234), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_create( |
| XPaths.scaling_group_instance(nsr1_uuid, "group", 12345), |
| NsrYang.YangData_Nsr_NsInstanceConfig_Nsr_ScalingGroup_Instance(id=12345), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| group_ids = self.handler.get_scale_group_instances(nsr2_uuid, "group") |
| self.log.debug("Got group ids in nsr2 after adding 12345 to nsr1: %s", group_ids) |
| group_ids = self.handler.get_scale_group_instances(nsr1_uuid, "group") |
| self.log.debug("Got group ids in nsr1 after adding 12345 to nsr1: %s", group_ids) |
| assert group_ids == {12345} |
| |
| self.log.debug("\n\nADD A COMPLETELY DIFFERENT NSR\n") |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_update( |
| XPaths.nsr_config(nsr2_uuid), |
| NsrYang.YangData_Nsr_NsInstanceConfig_Nsr(id=nsr2_uuid, name="fu2"), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| group_ids = self.handler.get_scale_group_instances(nsr2_uuid, "group") |
| self.log.debug("Got group ids in nsr2 after adding new nsr: %s", group_ids) |
| group_ids = self.handler.get_scale_group_instances(nsr1_uuid, "group") |
| self.log.debug("Got group ids in nsr1 after adding new nsr: %s", group_ids) |
| assert group_ids == {12345} |
| |
| self.log.debug("\n\nDELETE A COMPLETELY DIFFERENT NSR\n") |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_delete( |
| XPaths.nsr_config(nsr2_uuid), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(.5, loop=self.loop) |
| |
| group_ids = self.handler.get_scale_group_instances(nsr2_uuid, "group") |
| self.log.debug("Got group ids in nsr2 after deleting nsr2: %s", group_ids) |
| group_ids = self.handler.get_scale_group_instances(nsr1_uuid, "group") |
| self.log.debug("Got group ids in nsr1 after deleting nsr2: %s", group_ids) |
| assert group_ids == {12345} |
| |
| with self.dts_c.transaction() as xact: |
| block = xact.block_create() |
| block.add_query_delete( |
| XPaths.scaling_group_instance(nsr1_uuid, "group", 12345), |
| flags=rwdts.XactFlag.ADVISE | rwdts.XactFlag.TRACE, |
| ) |
| yield from block.execute(now=True) |
| |
| yield from asyncio.sleep(2, loop=self.loop) |
| |
| def main(): |
| runner = xmlrunner.XMLTestRunner(output=os.environ["RIFT_MODULE_TEST"]) |
| |
| parser = argparse.ArgumentParser() |
| parser.add_argument('-v', '--verbose', action='store_true') |
| parser.add_argument('-n', '--no-runner', action='store_true') |
| args, unittest_args = parser.parse_known_args() |
| if args.no_runner: |
| runner = None |
| |
| NsrHandlerTestCase.log_level = logging.DEBUG if args.verbose else logging.WARN |
| |
| unittest.main(testRunner=runner, argv=[sys.argv[0]] + unittest_args) |
| |
| if __name__ == '__main__': |
| main() |