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 from __future__ import absolute_import
  10
  11 import logging
  12 import os
  13 import time
  14
  15 from oslo_serialization import jsonutils
  16 import requests
  17
  18 from yardstick.benchmark.scenarios import base
  19
  20
  21 LOG = logging.getLogger(__name__)
  22
  23
  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
  30
  31   Command line options:
  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.
  37
  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
  46
  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.
  50
  51     """
  52     __scenario_type__ = "StorPerf"
  53
  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"]
  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 = jsonutils.loads(
  76             setup_query.content)
  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"])
  81             return True
  82
  83         return False
  84
  85     def setup(self):
  86         """Set the configuration."""
  87         env_args = {}
  88         env_args_payload_list = ["agent_count", "agent_flavor",
  89                                  "public_network", "agent_image",
  90                                  "volume_size", "volume_type",
  91                                  "volume_count", "availability_zone",
  92                                  "stack_name", "subnet_CIDR"]
  93
  94         for env_argument in env_args_payload_list:
  95             try:
  96                 env_args[env_argument] = self.options[env_argument]
  97             except KeyError:
  98                 pass
  99
 100         LOG.info("Creating a stack on node %s with parameters %s",
 101                  self.target, env_args)
 102         setup_res = requests.post('http://%s:5000/api/v1.0/configurations'
 103                                   % self.target, json=env_args)
 104
 105         setup_res_content = jsonutils.loads(
 106             setup_res.content)
 107
 108         if setup_res.status_code != 200:
 109             raise RuntimeError("Failed to create a stack, error message:",
 110                                setup_res_content["message"])
 111         elif setup_res.status_code == 200:
 112             LOG.info("stack_id: %s", setup_res_content["stack_id"])
 113
 114         while not self._query_setup_state():
 115             time.sleep(self.query_interval)
 116
 117         # We do not want to load the results of the disk initialization,
 118         # so it is not added to the results here.
 119         self.initialize_disks()
 120         self.setup_done = True
 121
 122     def _query_job_state(self, job_id):
 123         """Query the status of the supplied job_id and report on metrics"""
 124         LOG.info("Fetching report for %s...", job_id)
 125         report_res = requests.get('http://{}:5000/api/v1.0/jobs'.format
 126                                   (self.target),
 127                                   params={'id': job_id, 'type': 'status'})
 128
 129         report_res_content = jsonutils.loads(
 130             report_res.content)
 131
 132         if report_res.status_code != 200:
 133             raise RuntimeError("Failed to fetch report, error message:",
 134                                report_res_content["message"])
 135         else:
 136             job_status = report_res_content["Status"]
 137
 138         LOG.debug("Job is: %s...", job_status)
 139         self.job_completed = job_status == "Completed"
 140
 141         # TODO: Support using StorPerf ReST API to read Job ETA.
 142
 143         # if job_status == "completed":
 144         #     self.job_completed = True
 145         #     ETA = 0
 146         # elif job_status == "running":
 147         #     ETA = report_res_content['time']
 148         #
 149         # return ETA
 150
 151     def run(self, result):
 152         """Execute StorPerf benchmark"""
 153         if not self.setup_done:
 154             self.setup()
 155
 156         metadata = {"build_tag": "latest",
 157                     "test_case": "opnfv_yardstick_tc074"}
 158         metadata_payload_dict = {"pod_name": "NODE_NAME",
 159                                  "scenario_name": "DEPLOY_SCENARIO",
 160                                  "version": "YARDSTICK_BRANCH"}
 161
 162         for key, value in metadata_payload_dict.items():
 163             try:
 164                 metadata[key] = os.environ[value]
 165             except KeyError:
 166                 pass
 167
 168         job_args = {"metadata": metadata}
 169         job_args_payload_list = ["block_sizes", "queue_depths", "deadline",
 170                                  "target", "workload", "workloads",
 171                                  "agent_count", "steady_state_samples"]
 172         job_args["deadline"] = self.options["timeout"]
 173
 174         for job_argument in job_args_payload_list:
 175             try:
 176                 job_args[job_argument] = self.options[job_argument]
 177             except KeyError:
 178                 pass
 179
 180         api_version = "v1.0"
 181
 182         if ("workloads" in job_args and
 183                 job_args["workloads"] is not None and
 184                 len(job_args["workloads"])) > 0:
 185             api_version = "v2.0"
 186
 187         LOG.info("Starting a job with parameters %s", job_args)
 188         job_res = requests.post('http://%s:5000/api/%s/jobs' % (self.target,
 189                                                                 api_version),
 190                                 json=job_args)
 191
 192         job_res_content = jsonutils.loads(job_res.content)
 193
 194         if job_res.status_code != 200:
 195             raise RuntimeError("Failed to start a job, error message:",
 196                                job_res_content["message"])
 197         elif job_res.status_code == 200:
 198             job_id = job_res_content["job_id"]
 199             LOG.info("Started job id: %s...", job_id)
 200
 201             while not self.job_completed:
 202                 self._query_job_state(job_id)
 203                 time.sleep(self.query_interval)
 204
 205         # TODO: Support using ETA to polls for completion.
 206         #       Read ETA, next poll in 1/2 ETA time slot.
 207         #       If ETA is greater than the maximum allowed job time,
 208         #       then terminate job immediately.
 209
 210         #   while not self.job_completed:
 211         #       esti_time = self._query_state(job_id)
 212         #       if esti_time > self.timeout:
 213         #           terminate_res = requests.delete('http://%s:5000/api/v1.0
 214         #                                           /jobs' % self.target)
 215         #       else:
 216         #           time.sleep(int(esti_time)/2)
 217
 218             result_res = requests.get('http://%s:5000/api/v1.0/jobs?type='
 219                                       'metadata&id=%s' % (self.target, job_id))
 220             result_res_content = jsonutils.loads(result_res.content)
 221             if 'report' in result_res_content and \
 222                     'steady_state' in result_res_content['report']['details']:
 223                 res = result_res_content['report']['details']['steady_state']
 224                 steady_state = res.values()[0]
 225                 LOG.info("Job %s completed with steady state %s",
 226                          job_id, steady_state)
 227
 228             result_res = requests.get('http://%s:5000/api/v1.0/jobs?'
 229                                       'type=status&id=%s' % (self.target, job_id))
 230             result_res_content = jsonutils.loads(
 231                 result_res.content)
 232             result.update(result_res_content)
 233
 234     def initialize_disks(self):
 235         """Fills the target with random data prior to executing workloads"""
 236
 237         job_args = {}
 238         job_args_payload_list = ["target"]
 239
 240         for job_argument in job_args_payload_list:
 241             try:
 242                 job_args[job_argument] = self.options[job_argument]
 243             except KeyError:
 244                 pass
 245
 246         LOG.info("Starting initialization with parameters %s", job_args)
 247         job_res = requests.post('http://%s:5000/api/v1.0/initializations' %
 248                                 self.target, json=job_args)
 249
 250         job_res_content = jsonutils.loads(job_res.content)
 251
 252         if job_res.status_code != 200:
 253             raise RuntimeError(
 254                 "Failed to start initialization job, error message:",
 255                 job_res_content["message"])
 256         elif job_res.status_code == 200:
 257             job_id = job_res_content["job_id"]
 258             LOG.info("Started initialization as job id: %s...", job_id)
 259
 260         while not self.job_completed:
 261             self._query_job_state(job_id)
 262             time.sleep(self.query_interval)
 263
 264         self.job_completed = False
 265
 266     def teardown(self):
 267         """Deletes the agent configuration and the stack"""
 268         teardown_res = requests.delete(
 269             'http://%s:5000/api/v1.0/configurations' % self.target)
 270
 271         if teardown_res.status_code == 400:
 272             teardown_res_content = jsonutils.loads(
 273                 teardown_res.json_data)
 274             raise RuntimeError("Failed to reset environment, error message:",
 275                                teardown_res_content['message'])
 276
 277         self.setup_done = False