1 ##############################################################################
2 # Copyright (c) 2016 Huawei Technologies Co.,Ltd.
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
9 from __future__ import absolute_import
15 from oslo_serialization import jsonutils
18 from yardstick.benchmark.scenarios import base
21 LOG = logging.getLogger(__name__)
24 class StorPerf(base.Scenario):
25 """Execute StorPerf benchmark.
26 Once the StorPerf container has been started and the ReST API exposed,
27 you can interact directly with it using the ReST API. StorPerf comes with a
28 Swagger interface that is accessible through the exposed port at:
29 http://StorPerf:5000/swagger/index.html
32 target = [device or path] (Optional):
33 The path to either an attached storage device (/dev/vdb, etc) or a
34 directory path (/opt/storperf) that will be used to execute the performance
35 test. In the case of a device, the entire device will be used.
36 If not specified, the current directory will be used.
38 workload = [workload module] (Optional):
39 If not specified, the default is to run all workloads.
40 The workload types are:
41 rs: 100% Read, sequential data
42 ws: 100% Write, sequential data
43 rr: 100% Read, random access
44 wr: 100% Write, random access
45 rw: 70% Read / 30% write, random access
47 report = [job_id] (Optional):
48 Query the status of the supplied job_id and report on metrics.
49 If a workload is supplied, will report on only that subset.
52 __scenario_type__ = "StorPerf"
54 def __init__(self, scenario_cfg, context_cfg):
55 """Scenario construction."""
56 super(StorPerf, self).__init__()
57 self.scenario_cfg = scenario_cfg
58 self.context_cfg = context_cfg
59 self.options = self.scenario_cfg["options"]
61 self.target = self.options.get("StorPerf_ip", None)
62 self.query_interval = self.options.get("query_interval", 10)
63 # Maximum allowed job time
64 self.timeout = self.options.get('timeout', 3600)
66 self.setup_done = False
67 self.job_completed = False
69 def _query_setup_state(self):
70 """Query the stack status."""
71 LOG.info("Querying the stack state...")
72 setup_query = requests.get('http://%s:5000/api/v1.0/configurations'
75 setup_query_content = jsonutils.loads(
77 if ("stack_created" in setup_query_content and
78 setup_query_content["stack_created"]):
79 LOG.debug("stack_created: %s",
80 setup_query_content["stack_created"])
86 """Set the configuration."""
88 env_args_payload_list = ["agent_count", "agent_flavor",
89 "public_network", "agent_image",
92 for env_argument in env_args_payload_list:
94 env_args[env_argument] = self.options[env_argument]
98 LOG.info("Creating a stack on node %s with parameters %s",
99 self.target, env_args)
100 setup_res = requests.post('http://%s:5000/api/v1.0/configurations'
101 % self.target, json=env_args)
103 setup_res_content = jsonutils.loads(
106 if setup_res.status_code != 200:
107 raise RuntimeError("Failed to create a stack, error message:",
108 setup_res_content["message"])
109 elif setup_res.status_code == 200:
110 LOG.info("stack_id: %s", setup_res_content["stack_id"])
112 while not self._query_setup_state():
113 time.sleep(self.query_interval)
115 # We do not want to load the results of the disk initialization,
116 # so it is not added to the results here.
117 self.initialize_disks()
118 self.setup_done = True
120 def _query_job_state(self, job_id):
121 """Query the status of the supplied job_id and report on metrics"""
122 LOG.info("Fetching report for %s...", job_id)
123 report_res = requests.get('http://{}:5000/api/v1.0/jobs'.format
125 params={'id': job_id, 'type': 'status'})
127 report_res_content = jsonutils.loads(
130 if report_res.status_code != 200:
131 raise RuntimeError("Failed to fetch report, error message:",
132 report_res_content["message"])
134 job_status = report_res_content["Status"]
136 LOG.debug("Job is: %s...", job_status)
137 self.job_completed = job_status == "Completed"
139 # TODO: Support using StorPerf ReST API to read Job ETA.
141 # if job_status == "completed":
142 # self.job_completed = True
144 # elif job_status == "running":
145 # ETA = report_res_content['time']
149 def run(self, result):
150 """Execute StorPerf benchmark"""
151 if not self.setup_done:
154 metadata = {"build_tag": "latest",
155 "test_case": "opnfv_yardstick_tc074"}
156 metadata_payload_dict = {"pod_name": "NODE_NAME",
157 "scenario_name": "DEPLOY_SCENARIO",
158 "version": "YARDSTICK_BRANCH"}
160 for key, value in metadata_payload_dict.items():
162 metadata[key] = os.environ[value]
166 job_args = {"metadata": metadata}
167 job_args_payload_list = ["block_sizes", "queue_depths", "deadline",
168 "target", "workload", "workloads",
169 "agent_count", "steady_state_samples"]
170 job_args["deadline"] = self.options["timeout"]
172 for job_argument in job_args_payload_list:
174 job_args[job_argument] = self.options[job_argument]
180 if ("workloads" in job_args and
181 job_args["workloads"] is not None and
182 len(job_args["workloads"])) > 0:
185 LOG.info("Starting a job with parameters %s", job_args)
186 job_res = requests.post('http://%s:5000/api/%s/jobs' % (self.target,
190 job_res_content = jsonutils.loads(job_res.content)
192 if job_res.status_code != 200:
193 raise RuntimeError("Failed to start a job, error message:",
194 job_res_content["message"])
195 elif job_res.status_code == 200:
196 job_id = job_res_content["job_id"]
197 LOG.info("Started job id: %s...", job_id)
199 while not self.job_completed:
200 self._query_job_state(job_id)
201 time.sleep(self.query_interval)
203 # TODO: Support using ETA to polls for completion.
204 # Read ETA, next poll in 1/2 ETA time slot.
205 # If ETA is greater than the maximum allowed job time,
206 # then terminate job immediately.
208 # while not self.job_completed:
209 # esti_time = self._query_state(job_id)
210 # if esti_time > self.timeout:
211 # terminate_res = requests.delete('http://%s:5000/api/v1.0
212 # /jobs' % self.target)
214 # time.sleep(int(esti_time)/2)
216 result_res = requests.get('http://%s:5000/api/v1.0/jobs?id=%s' %
217 (self.target, job_id))
218 result_res_content = jsonutils.loads(
221 result.update(result_res_content)
223 def initialize_disks(self):
224 """Fills the target with random data prior to executing workloads"""
227 job_args_payload_list = ["target"]
229 for job_argument in job_args_payload_list:
231 job_args[job_argument] = self.options[job_argument]
235 LOG.info("Starting initialization with parameters %s", job_args)
236 job_res = requests.post('http://%s:5000/api/v1.0/initializations' %
237 self.target, json=job_args)
239 job_res_content = jsonutils.loads(job_res.content)
241 if job_res.status_code != 200:
243 "Failed to start initialization job, error message:",
244 job_res_content["message"])
245 elif job_res.status_code == 200:
246 job_id = job_res_content["job_id"]
247 LOG.info("Started initialization as job id: %s...", job_id)
249 while not self.job_completed:
250 self._query_job_state(job_id)
251 time.sleep(self.query_interval)
253 self.job_completed = False
256 """Deletes the agent configuration and the stack"""
257 teardown_res = requests.delete(
258 'http://%s:5000/api/v1.0/configurations' % self.target)
260 if teardown_res.status_code == 400:
261 teardown_res_content = jsonutils.loads(
262 teardown_res.json_data)
263 raise RuntimeError("Failed to reset environment, error message:",
264 teardown_res_content['message'])
266 self.setup_done = False