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         self.scenario_cfg = scenario_cfg
  58         self.context_cfg = context_cfg
  59         self.options = self.scenario_cfg["options"]
  60
  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)
  65
  66         self.setup_done = False
  67         self.job_completed = False
  68
  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'
  73                                    % self.target)
  74
  75         setup_query_content = json.loads(setup_query.content)
  76         if setup_query_content["stack_created"]:
  77             self.setup_done = True
  78             LOG.debug("stack_created: %s",
  79                       setup_query_content["stack_created"])
  80
  81     def setup(self):
  82         """Set the configuration."""
  83         env_args = {}
  84         env_args_payload_list = ["agent_count", "public_network",
  85                                  "agent_image", "volume_size"]
  86
  87         for env_argument in env_args_payload_list:
  88             if env_argument in self.options:
  89                 env_args[env_argument] = self.options[env_argument]
  90
  91         LOG.info("Creating a stack on node %s with parameters %s",
  92                  self.target, env_args)
  93         setup_res = requests.post('http://%s:5000/api/v1.0/configurations'
  94                                   % self.target, json=env_args)
  95
  96         setup_res_content = json.loads(setup_res.content)
  97
  98         if setup_res.status_code == 400:
  99             raise RuntimeError("Failed to create a stack, error message:",
 100                                setup_res_content["message"])
 101         elif setup_res.status_code == 200:
 102             LOG.info("stack_id: %s", setup_res_content["stack_id"])
 103
 104             while not self.setup_done:
 105                 self._query_setup_state()
 106                 time.sleep(self.query_interval)
 107
 108     # TODO: Support Storperf job status.
 109
 110     # def _query_job_state(self, job_id):
 111     #     """Query the status of the supplied job_id and report on metrics"""
 112     #     LOG.info("Fetching report for %s..." % job_id)
 113     #     report_res = requests.get('http://%s:5000/api/v1.0/jobs?id=%s' %
 114     #                               (self.target, job_id))
 115
 116     #     report_res_content = json.loads(report_res.content)
 117
 118     #     if report_res.status_code == 400:
 119     #         raise RuntimeError("Failed to fetch report, error message:",
 120     #                            report_res_content["message"])
 121     #     else:
 122     #         job_status = report_res_content["status"]
 123
 124     #     LOG.debug("Job is: %s..." % job_status)
 125     #     if job_status == "completed":
 126     #         self.job_completed = True
 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             if job_argument in self.options:
 149                 job_args[job_argument] = self.options[job_argument]
 150
 151         LOG.info("Starting a job with parameters %s", job_args)
 152         job_res = requests.post('http://%s:5000/api/v1.0/jobs' % self.target,
 153                                 json=job_args)
 154
 155         job_res_content = json.loads(job_res.content)
 156
 157         if job_res.status_code == 400:
 158             raise RuntimeError("Failed to start a job, error message:",
 159                                job_res_content["message"])
 160         elif job_res.status_code == 200:
 161             job_id = job_res_content["job_id"]
 162             LOG.info("Started job id: %s...", job_id)
 163
 164             time.sleep(self.timeout)
 165             terminate_res = requests.delete('http://%s:5000/api/v1.0/jobs' %
 166                                             self.target)
 167
 168             if terminate_res.status_code == 400:
 169                 terminate_res_content = json.loads(terminate_res.content)
 170                 raise RuntimeError("Failed to start a job, error message:",
 171                                    terminate_res_content["message"])
 172
 173         # TODO: Support Storperf job status.
 174
 175         #   while not self.job_completed:
 176         #       self._query_job_state(job_id)
 177         #       time.sleep(self.query_interval)
 178
 179         # TODO: Support using ETA to polls for completion.
 180         #       Read ETA, next poll in 1/2 ETA time slot.
 181         #       If ETA is greater than the maximum allowed job time,
 182         #       then terminate job immediately.
 183
 184         #   while not self.job_completed:
 185         #       esti_time = self._query_state(job_id)
 186         #       if esti_time > self.timeout:
 187         #           terminate_res = requests.delete('http://%s:5000/api/v1.0
 188         #                                           /jobs' % self.target)
 189         #       else:
 190         #           time.sleep(int(est_time)/2)
 191
 192             result_res = requests.get('http://%s:5000/api/v1.0/jobs?id=%s' %
 193                                       (self.target, job_id))
 194             result_res_content = json.loads(result_res.content)
 195
 196             result.update(result_res_content)
 197
 198     def teardown(self):
 199         """Deletes the agent configuration and the stack"""
 200         teardown_res = requests.delete('http://%s:5000/api/v1.0/\
 201                                        configurations' % self.target)
 202
 203         if teardown_res.status_code == 400:
 204             teardown_res_content = json.loads(teardown_res.content)
 205             raise RuntimeError("Failed to reset environment, error message:",
 206                                teardown_res_content['message'])
 207
 208         self.setup_done = False