--- /dev/null
+#!/usr/bin/env python
+
+##############################################################################
+# Copyright (c) 2016 Huawei Technologies Co.,Ltd.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+
+# Unittest for yardstick.benchmark.scenarios.storage.storperf.StorPerf
+
+import mock
+import unittest
+import requests
+import json
+
+from yardstick.benchmark.scenarios.storage import storperf
+
+
+def mocked_requests_config_post(*args, **kwargs):
+ class MockResponseConfigPost:
+ def __init__(self, json_data, status_code):
+ self.content = json_data
+ self.status_code = status_code
+
+ return MockResponseConfigPost('{"stack_id": "dac27db1-3502-4300-b301-91c64e6a1622","stack_created": "false"}', 200)
+
+
+def mocked_requests_config_get(*args, **kwargs):
+ class MockResponseConfigGet:
+ def __init__(self, json_data, status_code):
+ self.content = json_data
+ self.status_code = status_code
+
+ return MockResponseConfigGet('{"stack_id": "dac27db1-3502-4300-b301-91c64e6a1622","stack_created": "true"}', 200)
+
+
+def mocked_requests_job_get(*args, **kwargs):
+ class MockResponseJobGet:
+ def __init__(self, json_data, status_code):
+ self.content = json_data
+ self.status_code = status_code
+
+ return MockResponseJobGet('{"_ssd_preconditioning.queue-depth.8.block-size.16384.duration": 6}', 200)
+
+
+def mocked_requests_job_post(*args, **kwargs):
+ class MockResponseJobPost:
+ def __init__(self, json_data, status_code):
+ self.content = json_data
+ self.status_code = status_code
+
+ return MockResponseJobPost('{"job_id": \
+ "d46bfb8c-36f4-4a40-813b-c4b4a437f728"}', 200)
+
+
+def mocked_requests_job_delete(*args, **kwargs):
+ class MockResponseJobDelete:
+ def __init__(self, json_data, status_code):
+ self.content = json_data
+ self.status_code = status_code
+
+ return MockResponseJobDelete('{}', 200)
+
+
+def mocked_requests_delete(*args, **kwargs):
+ class MockResponseDelete:
+ def __init__(self, json_data, status_code):
+ self.json_data = json_data
+ self.status_code = status_code
+
+ return MockResponseDelete('{}', 200)
+
+
+def mocked_requests_delete_failed(*args, **kwargs):
+ class MockResponseDeleteFailed:
+ def __init__(self, json_data, status_code):
+ self.json_data = json_data
+ self.status_code = status_code
+
+ if args[0] == "http://172.16.0.137:5000/api/v1.0/configurations":
+ return MockResponseDeleteFailed('{"message": "Teardown failed"}', 400)
+
+ return MockResponseDeleteFailed('{}', 404)
+
+
+class StorPerfTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.ctx = {
+ 'host': {
+ 'ip': '172.16.0.137',
+ 'user': 'cirros',
+ 'key_filename': "mykey.key"
+ }
+ }
+
+ self.result = {}
+
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.post',
+ side_effect=mocked_requests_config_post)
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.get',
+ side_effect=mocked_requests_config_get)
+ def test_successful_setup(self, mock_post, mock_get):
+ options = {
+ "agent_count": 8,
+ "public_network": 'ext-net',
+ "volume_size": 10,
+ "block_sizes": 4096,
+ "queue_depths": 4,
+ "workload": "rs",
+ "StorPerf_ip": "192.168.23.2",
+ "query_interval": 10,
+ "timeout": 60
+ }
+
+ args = {
+ "options": options
+ }
+
+ s = storperf.StorPerf(args, self.ctx)
+
+ s.setup()
+
+ self.assertTrue(s.setup_done)
+
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.post',
+ side_effect=mocked_requests_job_post)
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.get',
+ side_effect=mocked_requests_job_get)
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.delete',
+ side_effect=mocked_requests_job_delete)
+ def test_successful_run(self, mock_post, mock_get, mock_delete):
+ options = {
+ "agent_count": 8,
+ "public_network": 'ext-net',
+ "volume_size": 10,
+ "block_sizes": 4096,
+ "queue_depths": 4,
+ "workload": "rs",
+ "StorPerf_ip": "192.168.23.2",
+ "query_interval": 10,
+ "timeout": 60
+ }
+
+ args = {
+ "options": options
+ }
+
+ s = storperf.StorPerf(args, self.ctx)
+ s.setup_done = True
+
+ sample_output = '{"_ssd_preconditioning.queue-depth.8.block-size.16384.duration": 6}'
+
+ expected_result = json.loads(sample_output)
+
+ s.run(self.result)
+
+ self.assertEqual(self.result, expected_result)
+
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.delete', side_effect=mocked_requests_delete)
+ def test_successful_teardown(self, mock_delete):
+ options = {
+ "agent_count": 8,
+ "public_network": 'ext-net',
+ "volume_size": 10,
+ "block_sizes": 4096,
+ "queue_depths": 4,
+ "workload": "rs",
+ "StorPerf_ip": "192.168.23.2",
+ "query_interval": 10,
+ "timeout": 60
+ }
+
+ args = {
+ "options": options
+ }
+
+ s = storperf.StorPerf(args, self.ctx)
+
+ s.teardown()
+
+ self.assertFalse(s.setup_done)
+
+ @mock.patch('yardstick.benchmark.scenarios.storage.storperf.requests.delete', side_effect=mocked_requests_delete_failed)
+ def test_failed_teardown(self, mock_delete):
+ options = {
+ "agent_count": 8,
+ "public_network": 'ext-net',
+ "volume_size": 10,
+ "block_sizes": 4096,
+ "queue_depths": 4,
+ "workload": "rs",
+ "StorPerf_ip": "192.168.23.2",
+ "query_interval": 10,
+ "timeout": 60
+ }
+
+ args = {
+ "options": options
+ }
+
+ s = storperf.StorPerf(args, self.ctx)
+
+ self.assertRaises(AssertionError, s.teardown(), self.result)
+
+
+def main():
+ unittest.main()
+
+if __name__ == '__main__':
+ main()
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 Huawei Technologies Co.,Ltd.
+#
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+import logging
+import json
+import requests
+import time
+
+from yardstick.benchmark.scenarios import base
+
+LOG = logging.getLogger(__name__)
+
+
+class StorPerf(base.Scenario):
+ """Execute StorPerf benchmark.
+ Once the StorPerf container has been started and the ReST API exposed,
+ you can interact directly with it using the ReST API. StorPerf comes with a
+ Swagger interface that is accessible through the exposed port at:
+ http://StorPerf:5000/swagger/index.html
+
+ Command line options:
+ target = [device or path] (Optional):
+ The path to either an attached storage device (/dev/vdb, etc) or a
+ directory path (/opt/storperf) that will be used to execute the performance
+ test. In the case of a device, the entire device will be used.
+ If not specified, the current directory will be used.
+
+ workload = [workload module] (Optional):
+ If not specified, the default is to run all workloads.
+ The workload types are:
+ rs: 100% Read, sequential data
+ ws: 100% Write, sequential data
+ rr: 100% Read, random access
+ wr: 100% Write, random access
+ rw: 70% Read / 30% write, random access
+
+ nossd (Optional):
+ Do not perform SSD style preconditioning.
+
+ nowarm (Optional):
+ Do not perform a warmup prior to measurements.
+
+ report = [job_id] (Optional):
+ Query the status of the supplied job_id and report on metrics.
+ If a workload is supplied, will report on only that subset.
+
+ """
+ __scenario_type__ = "StorPerf"
+
+ def __init__(self, scenario_cfg, context_cfg):
+ """Scenario construction."""
+ self.scenario_cfg = scenario_cfg
+ self.context_cfg = context_cfg
+ self.options = self.scenario_cfg["options"]
+
+ self.target = self.options.get("StorPerf_ip", None)
+ self.query_interval = self.options.get("query_interval", 10)
+ # Maximum allowed job time
+ self.timeout = self.options.get('timeout', 3600)
+
+ self.setup_done = False
+ self.job_completed = False
+
+ def _query_setup_state(self):
+ """Query the stack status."""
+ LOG.info("Querying the stack state...")
+ setup_query = requests.get('http://%s:5000/api/v1.0/configurations'
+ % self.target)
+
+ setup_query_content = json.loads(setup_query.content)
+ if setup_query_content["stack_created"]:
+ self.setup_done = True
+ LOG.debug("stack_created: %s"
+ % setup_query_content["stack_created"])
+
+ def setup(self):
+ """Set the configuration."""
+ env_args = {}
+ env_args_payload_list = ["agent_count", "public_network",
+ "agent_image", "volume_size"]
+
+ for env_argument in env_args_payload_list:
+ if env_argument in self.options:
+ env_args[env_argument] = self.options[env_argument]
+
+ LOG.info("Creating a stack on node %s with parameters %s" %
+ (self.target, env_args))
+ setup_res = requests.post('http://%s:5000/api/v1.0/configurations'
+ % self.target, json=env_args)
+
+ setup_res_content = json.loads(setup_res.content)
+
+ if setup_res.status_code == 400:
+ raise RuntimeError("Failed to create a stack, error message:",
+ setup_res_content["message"])
+ elif setup_res.status_code == 200:
+ LOG.info("stack_id: %s" % setup_res_content["stack_id"])
+
+ while not self.setup_done:
+ self._query_setup_state()
+ time.sleep(self.query_interval)
+
+ # TODO: Support Storperf job status.
+
+ # def _query_job_state(self, job_id):
+ # """Query the status of the supplied job_id and report on metrics"""
+ # LOG.info("Fetching report for %s..." % job_id)
+ # report_res = requests.get('http://%s:5000/api/v1.0/jobs?id=%s' %
+ # (self.target, job_id))
+
+ # report_res_content = json.loads(report_res.content)
+
+ # if report_res.status_code == 400:
+ # raise RuntimeError("Failed to fetch report, error message:",
+ # report_res_content["message"])
+ # else:
+ # job_status = report_res_content["status"]
+
+ # LOG.debug("Job is: %s..." % job_status)
+ # if job_status == "completed":
+ # self.job_completed = True
+
+ # TODO: Support using StorPerf ReST API to read Job ETA.
+
+ # if job_status == "completed":
+ # self.job_completed = True
+ # ETA = 0
+ # elif job_status == "running":
+ # ETA = report_res_content['time']
+ #
+ # return ETA
+
+ def run(self, result):
+ """Execute StorPerf benchmark"""
+ if not self.setup_done:
+ self.setup()
+
+ job_args = {}
+ job_args_payload_list = ["block_sizes", "queue_depths", "deadline",
+ "target", "nossd", "nowarm", "workload"]
+
+ for job_argument in job_args_payload_list:
+ if job_argument in self.options:
+ job_args[job_argument] = self.options[job_argument]
+
+ LOG.info("Starting a job with parameters %s" % job_args)
+ job_res = requests.post('http://%s:5000/api/v1.0/jobs' % self.target,
+ json=job_args)
+
+ job_res_content = json.loads(job_res.content)
+
+ if job_res.status_code == 400:
+ raise RuntimeError("Failed to start a job, error message:",
+ job_res_content["message"])
+ elif job_res.status_code == 200:
+ job_id = job_res_content["job_id"]
+ LOG.info("Started job id: %s..." % job_id)
+
+ time.sleep(self.timeout)
+ terminate_res = requests.delete('http://%s:5000/api/v1.0/jobs' %
+ self.target)
+
+ if terminate_res.status_code == 400:
+ terminate_res_content = json.loads(terminate_res.content)
+ raise RuntimeError("Failed to start a job, error message:",
+ terminate_res_content["message"])
+
+ # TODO: Support Storperf job status.
+
+ # while not self.job_completed:
+ # self._query_job_state(job_id)
+ # time.sleep(self.query_interval)
+
+ # TODO: Support using ETA to polls for completion.
+ # Read ETA, next poll in 1/2 ETA time slot.
+ # If ETA is greater than the maximum allowed job time,
+ # then terminate job immediately.
+
+ # while not self.job_completed:
+ # esti_time = self._query_state(job_id)
+ # if esti_time > self.timeout:
+ # terminate_res = requests.delete('http://%s:5000/api/v1.0
+ # /jobs' % self.target)
+ # else:
+ # time.sleep(int(est_time)/2)
+
+ result_res = requests.get('http://%s:5000/api/v1.0/jobs?id=%s' %
+ (self.target, job_id))
+ result_res_content = json.loads(result_res.content)
+
+ result.update(result_res_content)
+
+ def teardown(self):
+ """Deletes the agent configuration and the stack"""
+ teardown_res = requests.delete('http://%s:5000/api/v1.0/\
+ configurations' % self.target)
+
+ if teardown_res.status_code == 400:
+ teardown_res_content = json.loads(teardown_res.content)
+ raise RuntimeError("Failed to reset environment, error message:",
+ teardown_res_content['message'])
+
+ self.setup_done = False
Code Review / yardstick.git / commitdiff