Merge "ssh.py: add flag to request for a pseudo terminal (pty) for ssh connection"
[yardstick.git] / yardstick / benchmark / scenarios / storage / storperf.py
1 ##############################################################################
2 # Copyright (c) 2016 Huawei Technologies Co.,Ltd.
3 #
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 import logging
10 import json
11 import requests
12 import time
13
14 from yardstick.benchmark.scenarios import base
15
16 LOG = logging.getLogger(__name__)
17
18
19 class StorPerf(base.Scenario):
20     """Execute StorPerf benchmark.
21     Once the StorPerf container has been started and the ReST API exposed,
22     you can interact directly with it using the ReST API. StorPerf comes with a
23     Swagger interface that is accessible through the exposed port at:
24     http://StorPerf:5000/swagger/index.html
25
26   Command line options:
27     target = [device or path] (Optional):
28     The path to either an attached storage device (/dev/vdb, etc) or a
29     directory path (/opt/storperf) that will be used to execute the performance
30     test. In the case of a device, the entire device will be used.
31     If not specified, the current directory will be used.
32
33     workload = [workload module] (Optional):
34     If not specified, the default is to run all workloads.
35     The workload types are:
36         rs: 100% Read, sequential data
37         ws: 100% Write, sequential data
38         rr: 100% Read, random access
39         wr: 100% Write, random access
40         rw: 70% Read / 30% write, random access
41
42     nossd (Optional):
43     Do not perform SSD style preconditioning.
44
45     nowarm (Optional):
46     Do not perform a warmup prior to measurements.
47
48     report = [job_id] (Optional):
49     Query the status of the supplied job_id and report on metrics.
50     If a workload is supplied, will report on only that subset.
51
52     """
53     __scenario_type__ = "StorPerf"
54
55     def __init__(self, scenario_cfg, context_cfg):
56         """Scenario construction."""
57         super(StorPerf, self).__init__()
58         self.scenario_cfg = scenario_cfg
59         self.context_cfg = context_cfg
60         self.options = self.scenario_cfg["options"]
61
62         self.target = self.options.get("StorPerf_ip", None)
63         self.query_interval = self.options.get("query_interval", 10)
64         # Maximum allowed job time
65         self.timeout = self.options.get('timeout', 3600)
66
67         self.setup_done = False
68         self.job_completed = False
69
70     def _query_setup_state(self):
71         """Query the stack status."""
72         LOG.info("Querying the stack state...")
73         setup_query = requests.get('http://%s:5000/api/v1.0/configurations'
74                                    % self.target)
75
76         setup_query_content = json.loads(setup_query.content)
77         if setup_query_content["stack_created"]:
78             self.setup_done = True
79             LOG.debug("stack_created: %s",
80                       setup_query_content["stack_created"])
81
82     def setup(self):
83         """Set the configuration."""
84         env_args = {}
85         env_args_payload_list = ["agent_count", "public_network",
86                                  "agent_image", "volume_size"]
87
88         for env_argument in env_args_payload_list:
89             try:
90                 env_args[env_argument] = self.options[env_argument]
91             except KeyError:
92                 pass
93
94         LOG.info("Creating a stack on node %s with parameters %s",
95                  self.target, env_args)
96         setup_res = requests.post('http://%s:5000/api/v1.0/configurations'
97                                   % self.target, json=env_args)
98
99         setup_res_content = json.loads(setup_res.content)
100
101         if setup_res.status_code != 200:
102             raise RuntimeError("Failed to create a stack, error message:",
103                                setup_res_content["message"])
104         elif setup_res.status_code == 200:
105             LOG.info("stack_id: %s", setup_res_content["stack_id"])
106
107             while not self.setup_done:
108                 self._query_setup_state()
109                 time.sleep(self.query_interval)
110
111     def _query_job_state(self, job_id):
112         """Query the status of the supplied job_id and report on metrics"""
113         LOG.info("Fetching report for %s...", job_id)
114         report_res = requests.get('http://{}:5000/api/v1.0/jobs'.format
115                                   (self.target), params={'id': job_id})
116
117         report_res_content = json.loads(report_res.content)
118
119         if report_res.status_code != 200:
120             raise RuntimeError("Failed to fetch report, error message:",
121                                report_res_content["message"])
122         else:
123             job_status = report_res_content["status"]
124
125         LOG.debug("Job is: %s...", job_status)
126         self.job_completed = job_status == "completed"
127
128         # TODO: Support using StorPerf ReST API to read Job ETA.
129
130         # if job_status == "completed":
131         #     self.job_completed = True
132         #     ETA = 0
133         # elif job_status == "running":
134         #     ETA = report_res_content['time']
135         #
136         # return ETA
137
138     def run(self, result):
139         """Execute StorPerf benchmark"""
140         if not self.setup_done:
141             self.setup()
142
143         job_args = {}
144         job_args_payload_list = ["block_sizes", "queue_depths", "deadline",
145                                  "target", "nossd", "nowarm", "workload"]
146
147         for job_argument in job_args_payload_list:
148             try:
149                 job_args[job_argument] = self.options[job_argument]
150             except KeyError:
151                 pass
152
153         LOG.info("Starting a job with parameters %s", job_args)
154         job_res = requests.post('http://%s:5000/api/v1.0/jobs' % self.target,
155                                 json=job_args)
156
157         job_res_content = json.loads(job_res.content)
158
159         if job_res.status_code != 200:
160             raise RuntimeError("Failed to start a job, error message:",
161                                job_res_content["message"])
162         elif job_res.status_code == 200:
163             job_id = job_res_content["job_id"]
164             LOG.info("Started job id: %s...", job_id)
165
166             while not self.job_completed:
167                 self._query_job_state(job_id)
168                 time.sleep(self.query_interval)
169
170             terminate_res = requests.delete('http://%s:5000/api/v1.0/jobs' %
171                                             self.target)
172
173             if terminate_res.status_code != 200:
174                 terminate_res_content = json.loads(terminate_res.content)
175                 raise RuntimeError("Failed to start a job, error message:",
176                                    terminate_res_content["message"])
177
178         # TODO: Support using ETA to polls for completion.
179         #       Read ETA, next poll in 1/2 ETA time slot.
180         #       If ETA is greater than the maximum allowed job time,
181         #       then terminate job immediately.
182
183         #   while not self.job_completed:
184         #       esti_time = self._query_state(job_id)
185         #       if esti_time > self.timeout:
186         #           terminate_res = requests.delete('http://%s:5000/api/v1.0
187         #                                           /jobs' % self.target)
188         #       else:
189         #           time.sleep(int(est_time)/2)
190
191             result_res = requests.get('http://%s:5000/api/v1.0/jobs?id=%s' %
192                                       (self.target, job_id))
193             result_res_content = json.loads(result_res.content)
194
195             result.update(result_res_content)
196
197     def teardown(self):
198         """Deletes the agent configuration and the stack"""
199         teardown_res = requests.delete('http://%s:5000/api/v1.0/\
200                                        configurations' % self.target)
201
202         if teardown_res.status_code == 400:
203             teardown_res_content = json.loads(teardown_res.content)
204             raise RuntimeError("Failed to reset environment, error message:",
205                                teardown_res_content['message'])
206
207         self.setup_done = False