yardstick/cmd/commands/task.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Ericsson AB and others.
   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
  10 """ Handler for yardstick command 'task' """
  11
  12 import sys
  13 import os
  14 import yaml
  15 import atexit
  16 import ipaddress
  17 import time
  18 import logging
  19 from yardstick.benchmark.contexts.base import Context
  20 from yardstick.benchmark.runners import base as base_runner
  21 from yardstick.common.task_template import TaskTemplate
  22 from yardstick.common.utils import cliargs
  23
  24 output_file_default = "/tmp/yardstick.out"
  25 test_cases_dir_default = "tests/opnfv/test_cases/"
  26 LOG = logging.getLogger(__name__)
  27
  28
  29 class TaskCommands(object):
  30     '''Task commands.
  31
  32        Set of commands to manage benchmark tasks.
  33     '''
  34
  35     @cliargs("inputfile", type=str, help="path to task or suite file", nargs=1)
  36     @cliargs("--task-args", dest="task_args",
  37              help="Input task args (dict in json). These args are used"
  38              "to render input task that is jinja2 template.")
  39     @cliargs("--task-args-file", dest="task_args_file",
  40              help="Path to the file with input task args (dict in "
  41              "json/yaml). These args are used to render input"
  42              "task that is jinja2 template.")
  43     @cliargs("--keep-deploy", help="keep context deployed in cloud",
  44              action="store_true")
  45     @cliargs("--parse-only", help="parse the config file and exit",
  46              action="store_true")
  47     @cliargs("--output-file", help="file where output is stored, default %s" %
  48              output_file_default, default=output_file_default)
  49     @cliargs("--suite", help="process test suite file instead of a task file",
  50              action="store_true")
  51     def do_start(self, args):
  52         '''Start a benchmark scenario.'''
  53
  54         atexit.register(atexit_handler)
  55
  56         total_start_time = time.time()
  57         parser = TaskParser(args.inputfile[0])
  58
  59         suite_params = {}
  60         if args.suite:
  61             suite_params = parser.parse_suite()
  62             test_cases_dir = suite_params["test_cases_dir"]
  63             if test_cases_dir[-1] != os.sep:
  64                 test_cases_dir += os.sep
  65             task_files = [test_cases_dir + task
  66                           for task in suite_params["task_fnames"]]
  67         else:
  68             task_files = [parser.path]
  69
  70         task_args = suite_params.get("task_args", [args.task_args])
  71         task_args_fnames = suite_params.get("task_args_fnames",
  72                                             [args.task_args_file])
  73
  74         if args.parse_only:
  75             sys.exit(0)
  76
  77         if os.path.isfile(args.output_file):
  78             os.remove(args.output_file)
  79
  80         for i in range(0, len(task_files)):
  81             one_task_start_time = time.time()
  82             parser.path = task_files[i]
  83             scenarios, run_in_parallel = parser.parse_task(task_args[i],
  84                                                            task_args_fnames[i])
  85
  86             self._run(scenarios, run_in_parallel, args.output_file)
  87
  88             if args.keep_deploy:
  89                 # keep deployment, forget about stack
  90                 # (hide it for exit handler)
  91                 Context.list = []
  92             else:
  93                 for context in Context.list:
  94                     context.undeploy()
  95                 Context.list = []
  96             one_task_end_time = time.time()
  97             LOG.info("task %s finished in %d secs", task_files[i],
  98                      one_task_end_time - one_task_start_time)
  99
 100         total_end_time = time.time()
 101         LOG.info("total finished in %d secs",
 102                  total_end_time - total_start_time)
 103
 104         print "Done, exiting"
 105
 106     def _run(self, scenarios, run_in_parallel, output_file):
 107         '''Deploys context and calls runners'''
 108         for context in Context.list:
 109             context.deploy()
 110
 111         runners = []
 112         if run_in_parallel:
 113             for scenario in scenarios:
 114                 runner = run_one_scenario(scenario, output_file)
 115                 runners.append(runner)
 116
 117             # Wait for runners to finish
 118             for runner in runners:
 119                 runner_join(runner)
 120                 print "Runner ended, output in", output_file
 121         else:
 122             # run serially
 123             for scenario in scenarios:
 124                 runner = run_one_scenario(scenario, output_file)
 125                 runner_join(runner)
 126                 print "Runner ended, output in", output_file
 127
 128 # TODO: Move stuff below into TaskCommands class !?
 129
 130
 131 class TaskParser(object):
 132     '''Parser for task config files in yaml format'''
 133     def __init__(self, path):
 134         self.path = path
 135
 136     def parse_suite(self):
 137         '''parse the suite file and return a list of task config file paths
 138            and lists of optional parameters if present'''
 139         print "Parsing suite file:", self.path
 140
 141         try:
 142             with open(self.path) as stream:
 143                 cfg = yaml.load(stream)
 144         except IOError as ioerror:
 145             sys.exit(ioerror)
 146
 147         self._check_schema(cfg["schema"], "suite")
 148         print "Starting suite:", cfg["name"]
 149
 150         test_cases_dir = cfg.get("test_cases_dir", test_cases_dir_default)
 151         task_fnames = []
 152         task_args = []
 153         task_args_fnames = []
 154
 155         for task in cfg["test_cases"]:
 156             task_fnames.append(task["file_name"])
 157             if "task_args" in task:
 158                 task_args.append(task["task_args"])
 159             else:
 160                 task_args.append(None)
 161
 162             if "task_args_file" in task:
 163                 task_args_fnames.append(task["task_args_file"])
 164             else:
 165                 task_args_fnames.append(None)
 166
 167         suite_params = {
 168             "test_cases_dir": test_cases_dir,
 169             "task_fnames": task_fnames,
 170             "task_args": task_args,
 171             "task_args_fnames": task_args_fnames
 172         }
 173
 174         return suite_params
 175
 176     def parse_task(self, task_args=None, task_args_file=None):
 177         '''parses the task file and return an context and scenario instances'''
 178         print "Parsing task config:", self.path
 179
 180         try:
 181             kw = {}
 182             if task_args_file:
 183                 with open(task_args_file) as f:
 184                     kw.update(parse_task_args("task_args_file", f.read()))
 185             kw.update(parse_task_args("task_args", task_args))
 186         except TypeError:
 187             raise TypeError()
 188
 189         try:
 190             with open(self.path) as f:
 191                 try:
 192                     input_task = f.read()
 193                     rendered_task = TaskTemplate.render(input_task, **kw)
 194                 except Exception as e:
 195                     print(("Failed to render template:\n%(task)s\n%(err)s\n")
 196                           % {"task": input_task, "err": e})
 197                     raise e
 198                 print(("Input task is:\n%s\n") % rendered_task)
 199
 200                 cfg = yaml.load(rendered_task)
 201         except IOError as ioerror:
 202             sys.exit(ioerror)
 203
 204         self._check_schema(cfg["schema"], "task")
 205
 206         # TODO: support one or many contexts? Many would simpler and precise
 207         # TODO: support hybrid context type
 208         if "context" in cfg:
 209             context_cfgs = [cfg["context"]]
 210         else:
 211             context_cfgs = cfg["contexts"]
 212
 213         for cfg_attrs in context_cfgs:
 214             context_type = cfg_attrs.get("type", "Heat")
 215             if "Heat" == context_type and "networks" in cfg_attrs:
 216                 # config external_network based on env var
 217                 for _, attrs in cfg_attrs["networks"].items():
 218                     attrs["external_network"] = os.environ.get(
 219                         'EXTERNAL_NETWORK', 'net04_ext')
 220             context = Context.get(context_type)
 221             context.init(cfg_attrs)
 222
 223         run_in_parallel = cfg.get("run_in_parallel", False)
 224
 225         # TODO we need something better here, a class that represent the file
 226         return cfg["scenarios"], run_in_parallel
 227
 228     def _check_schema(self, cfg_schema, schema_type):
 229         '''Check if config file is using the correct schema type'''
 230
 231         if cfg_schema != "yardstick:" + schema_type + ":0.1":
 232             sys.exit("error: file %s has unknown schema %s" % (self.path,
 233                                                                cfg_schema))
 234
 235
 236 def atexit_handler():
 237     '''handler for process termination'''
 238     base_runner.Runner.terminate_all()
 239
 240     if len(Context.list) > 0:
 241         print "Undeploying all contexts"
 242         for context in Context.list:
 243             context.undeploy()
 244
 245
 246 def is_ip_addr(addr):
 247     '''check if string addr is an IP address'''
 248     try:
 249         ipaddress.ip_address(unicode(addr))
 250         return True
 251     except ValueError:
 252         return False
 253
 254
 255 def _is_same_heat_context(host_attr, target_attr):
 256     '''check if two servers are in the same heat context
 257     host_attr: either a name for a server created by yardstick or a dict
 258     with attribute name mapping when using external heat templates
 259     target_attr: either a name for a server created by yardstick or a dict
 260     with attribute name mapping when using external heat templates
 261     '''
 262     host = None
 263     target = None
 264     for context in Context.list:
 265         if context.__context_type__ != "Heat":
 266             continue
 267
 268         host = context._get_server(host_attr)
 269         if host is None:
 270             continue
 271
 272         target = context._get_server(target_attr)
 273         if target is None:
 274             return False
 275
 276         # Both host and target is not None, then they are in the
 277         # same heat context.
 278         return True
 279
 280     return False
 281
 282
 283 def run_one_scenario(scenario_cfg, output_file):
 284     '''run one scenario using context'''
 285     runner_cfg = scenario_cfg["runner"]
 286     runner_cfg['output_filename'] = output_file
 287
 288     # TODO support get multi hosts/vms info
 289     context_cfg = {}
 290     context_cfg['host'] = Context.get_server(scenario_cfg["host"])
 291
 292     if "target" in scenario_cfg:
 293         if is_ip_addr(scenario_cfg["target"]):
 294             context_cfg['target'] = {}
 295             context_cfg['target']["ipaddr"] = scenario_cfg["target"]
 296         else:
 297             context_cfg['target'] = Context.get_server(scenario_cfg["target"])
 298             if _is_same_heat_context(scenario_cfg["host"],
 299                                      scenario_cfg["target"]):
 300                 context_cfg["target"]["ipaddr"] = \
 301                     context_cfg["target"]["private_ip"]
 302             else:
 303                 context_cfg["target"]["ipaddr"] = \
 304                     context_cfg["target"]["ip"]
 305
 306     runner = base_runner.Runner.get(runner_cfg)
 307
 308     print "Starting runner of type '%s'" % runner_cfg["type"]
 309     runner.run(scenario_cfg, context_cfg)
 310
 311     return runner
 312
 313
 314 def runner_join(runner):
 315     '''join (wait for) a runner, exit process at runner failure'''
 316     status = runner.join()
 317     base_runner.Runner.release(runner)
 318     if status != 0:
 319         sys.exit("Runner failed")
 320
 321
 322 def print_invalid_header(source_name, args):
 323     print(("Invalid %(source)s passed:\n\n %(args)s\n")
 324           % {"source": source_name, "args": args})
 325
 326
 327 def parse_task_args(src_name, args):
 328     try:
 329         kw = args and yaml.safe_load(args)
 330         kw = {} if kw is None else kw
 331     except yaml.parser.ParserError as e:
 332         print_invalid_header(src_name, args)
 333         print(("%(source)s has to be YAML. Details:\n\n%(err)s\n")
 334               % {"source": src_name, "err": e})
 335         raise TypeError()
 336
 337     if not isinstance(kw, dict):
 338         print_invalid_header(src_name, args)
 339         print(("%(src)s had to be dict, actually %(src_type)s\n")
 340               % {"src": src_name, "src_type": type(kw)})
 341         raise TypeError()
 342     return kw