--- /dev/null
+# Copyright (c) 2016-2017 Intel Corporation
+#
+# 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.
+
+---
+{% set provider = provider or none %}
+{% set physical_networks = physical_networks or ['physnet1', 'physnet2'] %}
+{% set segmentation_id = segmentation_id or none %}
+
+schema: yardstick:task:0.1
+scenarios:
+- type: NSPerf
+ traffic_profile: ../../traffic_profiles/ipv4_throughput.yaml
+ topology: vfw-tg-topology.yaml
+ nodes:
+ tg__0: trafficgen_1.yardstick
+ vnf__0: vnf.yardstick
+ options:
+ hugepages_gb: 8
+ framesize:
+ uplink: {64B: 100}
+ downlink: {64B: 100}
+ flow:
+ src_ip: [{'tg__0': 'xe0'}]
+ dst_ip: [{'tg__0': 'xe1'}]
+ count: 1
+ traffic_type: 4
+ rfc2544:
+ allowed_drop_rate: 0.0001 - 0.0001
+ vnf__0:
+ rules: acl_1rule.yaml
+ vnf_config: {lb_config: 'SW', lb_count: 1, worker_config: '1C/1T', worker_threads: 1}
+ runner:
+ type: IterationIPC
+ iterations: 10
+ timeout: 60
+context:
+ # put node context first, so we don't HEAT deploy if node has errors
+ name: yardstick
+ image: yardstick-samplevnfs
+ flavor:
+ vcpus: 10
+ ram: 12288
+ disk: 6
+ extra_specs:
+ hw:cpu_sockets: 1
+ hw:cpu_cores: 10
+ hw:cpu_threads: 1
+ user: ubuntu
+ placement_groups:
+ pgrp1:
+ policy: "availability"
+ servers:
+ vnf:
+ floating_ip: true
+ placement: "pgrp1"
+ trafficgen_1:
+ floating_ip: true
+ placement: "pgrp1"
+ networks:
+ mgmt:
+ cidr: '10.0.1.0/24'
+ xe0:
+ cidr: '10.0.2.0/24'
+ gateway_ip: 'null'
+ {% if provider %}
+ provider: {{ provider }}
+ physical_network: {{ physical_networks[0] }}
+ {% if segmentation_id %}
+ segmentation_id: {{ segmentation_id }}
+ {% endif %}
+ {% endif %}
+ port_security_enabled: False
+ enable_dhcp: 'false'
+ xe1:
+ cidr: '10.0.3.0/24'
+ gateway_ip: 'null'
+ {% if provider %}
+ provider: {{ provider }}
+ physical_network: {{ physical_networks[1] }}
+ {% if segmentation_id %}
+ segmentation_id: {{ segmentation_id }}
+ {% endif %}
+ {% endif %}
+ port_security_enabled: False
+ enable_dhcp: 'false'
"""Endpoint class for ``RunnerIterationIPCConsumer``"""
def tg_method_started(self, ctxt, **kwargs):
- if ctxt['id'] in self._ctx_pids:
+ if ctxt['id'] in self._ctx_ids:
self._queue.put(
{'id': ctxt['id'],
'action': messaging.TG_METHOD_STARTED,
QUEUE_PUT_TIMEOUT)
def tg_method_finished(self, ctxt, **kwargs):
- if ctxt['id'] in self._ctx_pids:
+ if ctxt['id'] in self._ctx_ids:
self._queue.put(
{'id': ctxt['id'],
'action': messaging.TG_METHOD_FINISHED,
kwargs)})
def tg_method_iteration(self, ctxt, **kwargs):
- if ctxt['id'] in self._ctx_pids:
+ if ctxt['id'] in self._ctx_ids:
self._queue.put(
{'id': ctxt['id'],
'action': messaging.TG_METHOD_ITERATION,
if 'setup' not in run_step:
raise exceptions.RunnerIterationIPCSetupActionNeeded()
- producer_ctxs = benchmark.setup()
+ benchmark.setup()
+ producer_ctxs = benchmark.get_mq_ids()
if not producer_ctxs:
raise exceptions.RunnerIterationIPCNoCtxs()
except TypeError:
dic[k] = v
return dic
+
+ def get_mq_ids(self): # pragma: no cover
+ """Return stored MQ producer IDs, if defined"""
+ pass
__scenario_type__ = "NSPerf"
- def __init__(self, scenario_cfg, context_cfg): # Yardstick API
+ def __init__(self, scenario_cfg, context_cfg): # pragma: no cover
super(NetworkServiceTestCase, self).__init__()
self.scenario_cfg = scenario_cfg
self.context_cfg = context_cfg
self.traffic_profile = None
self.node_netdevs = {}
self.bin_path = get_nsb_option('bin_path', '')
+ self._mq_ids = []
def _get_ip_flow_range(self, ip_start_range):
topology_yaml = vnfdgen.generate_vnfd(topology, topolgy_data)
self.topology = topology_yaml["nsd:nsd-catalog"]["nsd"][0]
- def _find_vnf_name_from_id(self, vnf_id):
+ def _find_vnf_name_from_id(self, vnf_id): # pragma: no cover
return next((vnfd["vnfd-id-ref"]
for vnfd in self.topology["constituent-vnfd"]
if vnf_id == vnfd["member-vnf-index"]), None)
- def _find_vnfd_from_vnf_idx(self, vnf_id):
+ def _find_vnfd_from_vnf_idx(self, vnf_id): # pragma: no cover
return next((vnfd
for vnfd in self.topology["constituent-vnfd"]
if vnf_id == vnfd["member-vnf-index"]), None)
@staticmethod
- def find_node_if(nodes, name, if_name, vld_id):
+ def find_node_if(nodes, name, if_name, vld_id): # pragma: no cover
try:
# check for xe0, xe1
intf = nodes[name]["interfaces"][if_name]
node0_if["peer_intf"] = node1_copy
node1_if["peer_intf"] = node0_copy
- def _update_context_with_topology(self):
+ def _update_context_with_topology(self): # pragma: no cover
for vnfd in self.topology["constituent-vnfd"]:
vnf_idx = vnfd["member-vnf-index"]
vnf_name = self._find_vnf_name_from_id(vnf_idx)
vnfd = self._find_vnfd_from_vnf_idx(vnf_idx)
self.context_cfg["nodes"][vnf_name].update(vnfd)
- def _generate_pod_yaml(self):
+ def _generate_pod_yaml(self): # pragma: no cover
context_yaml = os.path.join(LOG_DIR, "pod-{}.yaml".format(self.scenario_cfg['task_id']))
# convert OrderedDict to a list
# pod.yaml nodes is a list
explicit_start=True)
@staticmethod
- def _serialize_node(node):
+ def _serialize_node(node): # pragma: no cover
new_node = copy.deepcopy(node)
# name field is required
# remove context suffix
self._update_context_with_topology()
@classmethod
- def get_vnf_impl(cls, vnf_model_id):
+ def get_vnf_impl(cls, vnf_model_id): # pragma: no cover
""" Find the implementing class from vnf_model["vnf"]["name"] field
:param vnf_model_id: parsed vnfd model ID field
raise exceptions.IncorrectConfig(error_msg=message)
@staticmethod
- def create_interfaces_from_node(vnfd, node):
+ def create_interfaces_from_node(vnfd, node): # pragma: no cover
ext_intfs = vnfd["vdu"][0]["external-interface"] = []
# have to sort so xe0 goes first
for intf_name, intf in sorted(node['interfaces'].items()):
return vnfs
def setup(self):
- """ Setup infrastructure, provission VNFs & start traffic
-
- :return:
- """
+ """Setup infrastructure, provission VNFs & start traffic"""
# 1. Verify if infrastructure mapping can meet topology
self.map_topology_to_infrastructure()
# 1a. Load VNF models
for traffic_gen in traffic_runners:
LOG.info("Starting traffic on %s", traffic_gen.name)
traffic_gen.run_traffic(self.traffic_profile)
+ self._mq_ids.append(traffic_gen.get_mq_producer_id())
+
+ def get_mq_ids(self): # pragma: no cover
+ """Return stored MQ producer IDs"""
+ return self._mq_ids
def run(self, result): # yardstick API
""" Yardstick calls run() at intervals defined in the yaml and
LOG.exception("")
raise RuntimeError("Error in teardown")
- def pre_run_wait_time(self, time_seconds):
+ def pre_run_wait_time(self, time_seconds): # pragma: no cover
"""Time waited before executing the run method"""
time.sleep(time_seconds)
- def post_run_wait_time(self, time_seconds):
+ def post_run_wait_time(self, time_seconds): # pragma: no cover
"""Time waited after executing the run method"""
pass
class NotificationHandler(object):
"""Abstract class to define a endpoint object for a MessagingConsumer"""
- def __init__(self, _id, ctx_pids, queue):
+ def __init__(self, _id, ctx_ids, queue):
self._id = _id
- self._ctx_pids = ctx_pids
+ self._ctx_ids = ctx_ids
self._queue = queue
the messages published by a `MessagingNotifier`.
"""
- def __init__(self, topic, pids, endpoints, fanout=True):
+ def __init__(self, topic, ids, endpoints, fanout=True):
"""Init function.
:param topic: (string) MQ exchange topic
- :param pids: (list of int) list of PIDs of the processes implementing
+ :param ids: (list of int) list of IDs of the processes implementing
the MQ Notifier which will be in the message context
:param endpoints: (list of class) list of classes implementing the
methods (see `MessagingNotifier.send_message) used by
:returns: `MessagingConsumer` class object
"""
- self._pids = pids
+ self._ids = ids
self._endpoints = endpoints
self._transport = oslo_messaging.get_rpc_transport(
cfg.CONF, url=messaging.TRANSPORT_URL)
messages in a message queue.
"""
- def __init__(self, topic, pid=os.getpid(), fanout=True):
+ def __init__(self, topic, _id=os.getpid(), fanout=True):
"""Init function.
:param topic: (string) MQ exchange topic
- :param pid: (int) PID of the process implementing this MQ Notifier
+ :param id: (int) ID of the process implementing this MQ Notifier
:param fanout: (bool) MQ clients may request that a copy of the message
be delivered to all servers listening on a topic by
setting fanout to ``True``, rather than just one of them
:returns: `MessagingNotifier` class object
"""
self._topic = topic
- self._pid = pid
+ self._id = _id
self._fanout = fanout
self._transport = oslo_messaging.get_rpc_transport(
cfg.CONF, url=messaging.TRANSPORT_URL)
consumer endpoints
:param payload: (subclass `Payload`) payload content
"""
- self._notifier.cast({'pid': self._pid},
+ self._notifier.cast({'id': self._id},
method,
**payload.obj_to_dict())
+
+ @property
+ def id(self):
+ """Return MQ producer ID"""
+ return self._id
This message producer must be instantiated in the process created
"run_traffic" process.
"""
- def __init__(self, pid):
+ def __init__(self, _id):
super(TrafficGeneratorProducer, self).__init__(messaging.TOPIC_TG,
- pid=pid)
+ _id=_id)
def tg_method_started(self, version=1):
"""Send a message to inform the traffic generation has started"""
"""
pass
- def _setup_mq_producer(self, pid):
+ @staticmethod
+ def _setup_mq_producer(id):
"""Setup the TG MQ producer to send messages between processes
:return: (derived class from ``MessagingProducer``) MQ producer object
"""
- return TrafficGeneratorProducer(pid)
+ return TrafficGeneratorProducer(id)
+
+ def get_mq_producer_id(self):
+ """Return the MQ producer ID if initialized"""
+ if self._mq_producer:
+ return self._mq_producer.get_id()
LOG.info("%s TG Server is up and running.", self.APP_NAME)
return self._tg_process.exitcode
- def _traffic_runner(self, traffic_profile, mq_pid):
+ def _traffic_runner(self, traffic_profile, mq_id):
# always drop connections first thing in new processes
# so we don't get paramiko errors
self.ssh_helper.drop_connection()
LOG.info("Starting %s client...", self.APP_NAME)
- self._mq_producer = self._setup_mq_producer(mq_pid)
+ self._mq_producer = self._setup_mq_producer(mq_id)
self.resource_helper.run_traffic(traffic_profile, self._mq_producer)
def run_traffic(self, traffic_profile):
name = '{}-{}-{}-{}'.format(self.name, self.APP_NAME,
traffic_profile.__class__.__name__,
os.getpid())
- mq_pid = uuid.uuid1().int
- self._traffic_process = Process(name=name, target=self._traffic_runner,
- args=(traffic_profile, mq_pid))
+ self._traffic_process = Process(
+ name=name, target=self._traffic_runner,
+ args=(traffic_profile, uuid.uuid1().int))
self._traffic_process.start()
# Wait for traffic process to start
while self.resource_helper.client_started.value == 0:
if not self._traffic_process.is_alive():
break
- return mq_pid
-
def collect_kpi(self):
# check if the tg processes have exited
physical_node = Context.get_physical_node_from_server(
class DummyEndpoint(consumer.NotificationHandler):
def info(self, ctxt, **kwargs):
- if ctxt['pid'] in self._ctx_pids:
- self._queue.put('ID {}, data: {}, pid: {}'.format(
- self._id, kwargs['data'], ctxt['pid']))
+ if ctxt['id'] in self._ctx_ids:
+ self._queue.put('Nr {}, data: {}, id: {}'.format(
+ self._id, kwargs['data'], ctxt['id']))
class DummyConsumer(consumer.MessagingConsumer):
- def __init__(self, _id, ctx_pids, queue):
+ def __init__(self, _id, ctx_ids, queue):
self._id = _id
- endpoints = [DummyEndpoint(_id, ctx_pids, queue)]
- super(DummyConsumer, self).__init__(TOPIC, ctx_pids, endpoints)
+ endpoints = [DummyEndpoint(_id, ctx_ids, queue)]
+ super(DummyConsumer, self).__init__(TOPIC, ctx_ids, endpoints)
class DummyProducer(producer.MessagingProducer):
pass
-def _run_consumer(_id, ctx_pids, queue):
- _consumer = DummyConsumer(_id, ctx_pids, queue)
+def _run_consumer(_id, ctx_ids, queue):
+ _consumer = DummyConsumer(_id, ctx_ids, queue)
_consumer.start_rpc_server()
_consumer.wait()
num_consumers = 10
ctx_1 = 100001
ctx_2 = 100002
- producers = [DummyProducer(TOPIC, pid=ctx_1),
- DummyProducer(TOPIC, pid=ctx_2)]
+ producers = [DummyProducer(TOPIC, _id=ctx_1),
+ DummyProducer(TOPIC, _id=ctx_2)]
processes = []
for i in range(num_consumers):
output.append(output_queue.get(True, 1))
self.assertEqual(num_consumers * 4, len(output))
- msg_template = 'ID {}, data: {}, pid: {}'
+ msg_template = 'Nr {}, data: {}, id: {}'
for i in range(num_consumers):
for ctx in [ctx_1, ctx_2]:
for message in ['message 0', 'message 1']:
tgen.verify_traffic = lambda x: verified_dict
tgen.terminate = mock.Mock(return_value=True)
tgen.name = "tgen__1"
+ tgen.run_traffic.return_value = 'tg_id'
vnf = mock.Mock(autospec=GenericVNF)
vnf.runs_traffic = False
vnf.terminate = mock.Mock(return_value=True)
self.s.load_vnf_models = mock.Mock(return_value=self.s.vnfs)
self.s._fill_traffic_profile = \
mock.Mock(return_value=TRAFFIC_PROFILE)
- self.assertIsNone(self.s.setup())
def test_setup_exception(self):
with mock.patch("yardstick.ssh.SSH") as ssh:
)
self.assertEqual(self.s.topology, 'fake_nsd')
+ def test_get_mq_ids(self):
+ self.assertEqual(self.s._mq_ids, self.s.get_mq_ids())
+
def test_teardown(self):
vnf = mock.Mock(autospec=GenericVNF)
vnf.terminate = mock.Mock(return_value=True)
topic='test_topic', fanout=True, server=messaging.SERVER)
mock_RPCClient.assert_called_once_with('test_rpc_transport',
'test_Target')
+
+ def test_id(self):
+ with mock.patch.object(oslo_messaging, 'RPCClient'), \
+ mock.patch.object(oslo_messaging, 'get_rpc_transport'), \
+ mock.patch.object(oslo_messaging, 'Target'):
+ msg_producer = _MessagingProducer('topic', 'id_to_check')
+ self.assertEqual('id_to_check', msg_producer.id)
}
+class _DummyGenericTrafficGen(base.GenericTrafficGen): # pragma: no cover
+
+ def run_traffic(self, *args):
+ pass
+
+ def terminate(self):
+ pass
+
+ def collect_kpi(self):
+ pass
+
+ def instantiate(self, *args):
+ pass
+
+ def scale(self, flavor=''):
+ pass
+
+
class FileAbsPath(object):
def __init__(self, module_file):
super(FileAbsPath, self).__init__()
"scale, terminate")
self.assertEqual(msg, str(exc.exception))
+ def test_get_mq_producer_id(self):
+ vnfd = {'benchmark': {'kpi': mock.ANY},
+ 'vdu': [{'external-interface': 'ext_int'}]
+ }
+ tg = _DummyGenericTrafficGen('name', vnfd)
+ tg._mq_producer = mock.Mock()
+ tg._mq_producer.get_id.return_value = 'fake_id'
+ self.assertEqual('fake_id', tg.get_mq_producer_id())
+
class TrafficGeneratorProducerTestCase(unittest.TestCase):
@mock.patch.object(cfg, 'CONF')
def test__init(self, mock_config, mock_transport, mock_rpcclient,
mock_target):
- pid = uuid.uuid1().int
- tg_producer = base.TrafficGeneratorProducer(pid)
+ _id = uuid.uuid1().int
+ tg_producer = base.TrafficGeneratorProducer(_id)
mock_transport.assert_called_once_with(
mock_config, url='rabbit://yardstick:yardstick@localhost:5672/')
mock_target.assert_called_once_with(topic=messaging.TOPIC_TG,
fanout=True,
server=messaging.SERVER)
mock_rpcclient.assert_called_once_with('rpc_transport', 'rpc_target')
- self.assertEqual(pid, tg_producer._pid)
+ self.assertEqual(_id, tg_producer._id)
self.assertEqual(messaging.TOPIC_TG, tg_producer._topic)
@mock.patch.object(oslo_messaging, 'Target', return_value='rpc_target')
self.sut.ssh_helper.run = mock.Mock()
self.sut._traffic_runner = mock.Mock(return_value=0)
self.sut.resource_helper.client_started.value = 1
- result = self.sut.run_traffic(mock_traffic_profile)
+ self.sut.run_traffic(mock_traffic_profile)
self.sut._traffic_process.terminate()
- self.assertIsNotNone(result)
def test_terminate(self):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
Code Review / yardstick.git / commitdiff