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 import uuid
  20 from itertools import ifilter
  21
  22 from yardstick.benchmark.contexts.base import Context
  23 from yardstick.benchmark.runners import base as base_runner
  24 from yardstick.common.task_template import TaskTemplate
  25 from yardstick.common.utils import cliargs
  26
  27 output_file_default = "/tmp/yardstick.out"
  28 test_cases_dir_default = "tests/opnfv/test_cases/"
  29 LOG = logging.getLogger(__name__)
  30
  31
  32 class TaskCommands(object):
  33     '''Task commands.
  34
  35        Set of commands to manage benchmark tasks.
  36     '''
  37
  38     @cliargs("inputfile", type=str, help="path to task or suite file", nargs=1)
  39     @cliargs("--task-args", dest="task_args",
  40              help="Input task args (dict in json). These args are used"
  41              "to render input task that is jinja2 template.")
  42     @cliargs("--task-args-file", dest="task_args_file",
  43              help="Path to the file with input task args (dict in "
  44              "json/yaml). These args are used to render input"
  45              "task that is jinja2 template.")
  46     @cliargs("--keep-deploy", help="keep context deployed in cloud",
  47              action="store_true")
  48     @cliargs("--parse-only", help="parse the config file and exit",
  49              action="store_true")
  50     @cliargs("--output-file", help="file where output is stored, default %s" %
  51              output_file_default, default=output_file_default)
  52     @cliargs("--suite", help="process test suite file instead of a task file",
  53              action="store_true")
  54     def do_start(self, args):
  55         '''Start a benchmark scenario.'''
  56
  57         atexit.register(atexit_handler)
  58
  59         total_start_time = time.time()
  60         parser = TaskParser(args.inputfile[0])
  61
  62         if args.suite:
  63             # 1.parse suite, return suite_params info
  64             task_files, task_args, task_args_fnames = \
  65                 parser.parse_suite()
  66         else:
  67             task_files = [parser.path]
  68             task_args = [args.task_args]
  69             task_args_fnames = [args.task_args_file]
  70
  71         LOG.info("\ntask_files:%s, \ntask_args:%s, \ntask_args_fnames:%s",
  72                  task_files, task_args, task_args_fnames)
  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         # parse task_files
  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, meet_precondition = parser.parse_task(
  84                  task_args[i], task_args_fnames[i])
  85
  86             if not meet_precondition:
  87                 LOG.info("meet_precondition is %s, please check envrionment",
  88                          meet_precondition)
  89                 continue
  90
  91             self._run(scenarios, run_in_parallel, args.output_file)
  92
  93             if args.keep_deploy:
  94                 # keep deployment, forget about stack
  95                 # (hide it for exit handler)
  96                 Context.list = []
  97             else:
  98                 for context in Context.list:
  99                     context.undeploy()
 100                 Context.list = []
 101             one_task_end_time = time.time()
 102             LOG.info("task %s finished in %d secs", task_files[i],
 103                      one_task_end_time - one_task_start_time)
 104
 105         total_end_time = time.time()
 106         LOG.info("total finished in %d secs",
 107                  total_end_time - total_start_time)
 108
 109         print "Done, exiting"
 110
 111     def _run(self, scenarios, run_in_parallel, output_file):
 112         '''Deploys context and calls runners'''
 113         for context in Context.list:
 114             context.deploy()
 115
 116         background_runners = []
 117
 118         # Start all background scenarios
 119         for scenario in ifilter(_is_background_scenario, scenarios):
 120             scenario["runner"] = dict(type="Duration", duration=1000000000)
 121             runner = run_one_scenario(scenario, output_file)
 122             background_runners.append(runner)
 123
 124         runners = []
 125         if run_in_parallel:
 126             for scenario in scenarios:
 127                 if not _is_background_scenario(scenario):
 128                     runner = run_one_scenario(scenario, output_file)
 129                     runners.append(runner)
 130
 131             # Wait for runners to finish
 132             for runner in runners:
 133                 runner_join(runner)
 134                 print "Runner ended, output in", output_file
 135         else:
 136             # run serially
 137             for scenario in scenarios:
 138                 if not _is_background_scenario(scenario):
 139                     runner = run_one_scenario(scenario, output_file)
 140                     runner_join(runner)
 141                     print "Runner ended, output in", output_file
 142
 143         # Abort background runners
 144         for runner in background_runners:
 145             runner.abort()
 146
 147         # Wait for background runners to finish
 148         for runner in background_runners:
 149             if runner.join(timeout=60) is None:
 150                 # Nuke if it did not stop nicely
 151                 base_runner.Runner.terminate(runner)
 152                 runner_join(runner)
 153             else:
 154                 base_runner.Runner.release(runner)
 155             print "Background task ended"
 156
 157
 158 # TODO: Move stuff below into TaskCommands class !?
 159
 160
 161 class TaskParser(object):
 162     '''Parser for task config files in yaml format'''
 163     def __init__(self, path):
 164         self.path = path
 165
 166     def _meet_constraint(self, task, cur_pod, cur_installer):
 167         if "constraint" in task:
 168             constraint = task.get('constraint', None)
 169             if constraint is not None:
 170                 tc_fit_pod = constraint.get('pod', None)
 171                 tc_fit_installer = constraint.get('installer', None)
 172                 LOG.info("cur_pod:%s, cur_installer:%s,tc_constraints:%s",
 173                          cur_pod, cur_installer, constraint)
 174                 if cur_pod and tc_fit_pod and cur_pod not in tc_fit_pod:
 175                     return False
 176                 if cur_installer and tc_fit_installer and \
 177                         cur_installer not in tc_fit_installer:
 178                     return False
 179         return True
 180
 181     def _get_task_para(self, task, cur_pod):
 182         task_args = task.get('task_args', None)
 183         if task_args is not None:
 184             task_args = task_args.get(cur_pod, None)
 185         task_args_fnames = task.get('task_args_fnames', None)
 186         if task_args_fnames is not None:
 187             task_args_fnames = task_args_fnames.get(cur_pod, None)
 188         return task_args, task_args_fnames
 189
 190     def parse_suite(self):
 191         '''parse the suite file and return a list of task config file paths
 192            and lists of optional parameters if present'''
 193         LOG.info("\nParsing suite file:%s", self.path)
 194
 195         try:
 196             with open(self.path) as stream:
 197                 cfg = yaml.load(stream)
 198         except IOError as ioerror:
 199             sys.exit(ioerror)
 200
 201         self._check_schema(cfg["schema"], "suite")
 202         LOG.info("\nStarting scenario:%s", cfg["name"])
 203
 204         test_cases_dir = cfg.get("test_cases_dir", test_cases_dir_default)
 205         if test_cases_dir[-1] != os.sep:
 206             test_cases_dir += os.sep
 207
 208         cur_pod = os.environ.get('NODE_NAME', None)
 209         cur_installer = os.environ.get('INSTALLER_TYPE', None)
 210
 211         valid_task_files = []
 212         valid_task_args = []
 213         valid_task_args_fnames = []
 214
 215         for task in cfg["test_cases"]:
 216             # 1.check file_name
 217             if "file_name" in task:
 218                 task_fname = task.get('file_name', None)
 219                 if task_fname is None:
 220                     continue
 221             else:
 222                 continue
 223             # 2.check constraint
 224             if self._meet_constraint(task, cur_pod, cur_installer):
 225                 valid_task_files.append(test_cases_dir + task_fname)
 226             else:
 227                 continue
 228             # 3.fetch task parameters
 229             task_args, task_args_fnames = self._get_task_para(task, cur_pod)
 230             valid_task_args.append(task_args)
 231             valid_task_args_fnames.append(task_args_fnames)
 232
 233         return valid_task_files, valid_task_args, valid_task_args_fnames
 234
 235     def parse_task(self, task_args=None, task_args_file=None):
 236         '''parses the task file and return an context and scenario instances'''
 237         print "Parsing task config:", self.path
 238
 239         try:
 240             kw = {}
 241             if task_args_file:
 242                 with open(task_args_file) as f:
 243                     kw.update(parse_task_args("task_args_file", f.read()))
 244             kw.update(parse_task_args("task_args", task_args))
 245         except TypeError:
 246             raise TypeError()
 247
 248         try:
 249             with open(self.path) as f:
 250                 try:
 251                     input_task = f.read()
 252                     rendered_task = TaskTemplate.render(input_task, **kw)
 253                 except Exception as e:
 254                     print(("Failed to render template:\n%(task)s\n%(err)s\n")
 255                           % {"task": input_task, "err": e})
 256                     raise e
 257                 print(("Input task is:\n%s\n") % rendered_task)
 258
 259                 cfg = yaml.load(rendered_task)
 260         except IOError as ioerror:
 261             sys.exit(ioerror)
 262
 263         self._check_schema(cfg["schema"], "task")
 264         meet_precondition = self._check_precondition(cfg)
 265
 266         # TODO: support one or many contexts? Many would simpler and precise
 267         # TODO: support hybrid context type
 268         if "context" in cfg:
 269             context_cfgs = [cfg["context"]]
 270         elif "contexts" in cfg:
 271             context_cfgs = cfg["contexts"]
 272         else:
 273             context_cfgs = [{"type": "Dummy"}]
 274
 275         for cfg_attrs in context_cfgs:
 276             context_type = cfg_attrs.get("type", "Heat")
 277             if "Heat" == context_type and "networks" in cfg_attrs:
 278                 # config external_network based on env var
 279                 for _, attrs in cfg_attrs["networks"].items():
 280                     attrs["external_network"] = os.environ.get(
 281                         'EXTERNAL_NETWORK', 'net04_ext')
 282             context = Context.get(context_type)
 283             context.init(cfg_attrs)
 284
 285         run_in_parallel = cfg.get("run_in_parallel", False)
 286
 287         # add tc and task id for influxdb extended tags
 288         task_id = str(uuid.uuid4())
 289         for scenario in cfg["scenarios"]:
 290             task_name = os.path.splitext(os.path.basename(self.path))[0]
 291             scenario["tc"] = task_name
 292             scenario["task_id"] = task_id
 293
 294         # TODO we need something better here, a class that represent the file
 295         return cfg["scenarios"], run_in_parallel, meet_precondition
 296
 297     def _check_schema(self, cfg_schema, schema_type):
 298         '''Check if config file is using the correct schema type'''
 299
 300         if cfg_schema != "yardstick:" + schema_type + ":0.1":
 301             sys.exit("error: file %s has unknown schema %s" % (self.path,
 302                                                                cfg_schema))
 303
 304     def _check_precondition(self, cfg):
 305         '''Check if the envrionment meet the preconditon'''
 306
 307         if "precondition" in cfg:
 308             precondition = cfg["precondition"]
 309             installer_type = precondition.get("installer_type", None)
 310             deploy_scenarios = precondition.get("deploy_scenarios", None)
 311             tc_fit_pods = precondition.get("pod_name", None)
 312             installer_type_env = os.environ.get('INSTALL_TYPE', None)
 313             deploy_scenario_env = os.environ.get('DEPLOY_SCENARIO', None)
 314             pod_name_env = os.environ.get('NODE_NAME', None)
 315
 316             LOG.info("installer_type: %s, installer_type_env: %s",
 317                      installer_type, installer_type_env)
 318             LOG.info("deploy_scenarios: %s, deploy_scenario_env: %s",
 319                      deploy_scenarios, deploy_scenario_env)
 320             LOG.info("tc_fit_pods: %s, pod_name_env: %s",
 321                      tc_fit_pods, pod_name_env)
 322             if installer_type and installer_type_env:
 323                 if installer_type_env not in installer_type:
 324                     return False
 325             if deploy_scenarios and deploy_scenario_env:
 326                 deploy_scenarios_list = deploy_scenarios.split(',')
 327                 for deploy_scenario in deploy_scenarios_list:
 328                     if deploy_scenario_env.startswith(deploy_scenario):
 329                         return True
 330                 return False
 331             if tc_fit_pods and pod_name_env:
 332                 if pod_name_env not in tc_fit_pods:
 333                     return False
 334         return True
 335
 336
 337 def atexit_handler():
 338     '''handler for process termination'''
 339     base_runner.Runner.terminate_all()
 340
 341     if len(Context.list) > 0:
 342         print "Undeploying all contexts"
 343         for context in Context.list:
 344             context.undeploy()
 345
 346
 347 def is_ip_addr(addr):
 348     '''check if string addr is an IP address'''
 349     try:
 350         ipaddress.ip_address(unicode(addr))
 351         return True
 352     except ValueError:
 353         return False
 354
 355
 356 def _is_same_heat_context(host_attr, target_attr):
 357     '''check if two servers are in the same heat context
 358     host_attr: either a name for a server created by yardstick or a dict
 359     with attribute name mapping when using external heat templates
 360     target_attr: either a name for a server created by yardstick or a dict
 361     with attribute name mapping when using external heat templates
 362     '''
 363     host = None
 364     target = None
 365     for context in Context.list:
 366         if context.__context_type__ != "Heat":
 367             continue
 368
 369         host = context._get_server(host_attr)
 370         if host is None:
 371             continue
 372
 373         target = context._get_server(target_attr)
 374         if target is None:
 375             return False
 376
 377         # Both host and target is not None, then they are in the
 378         # same heat context.
 379         return True
 380
 381     return False
 382
 383
 384 def _is_background_scenario(scenario):
 385     if "run_in_background" in scenario:
 386         return scenario["run_in_background"]
 387     else:
 388         return False
 389
 390
 391 def run_one_scenario(scenario_cfg, output_file):
 392     '''run one scenario using context'''
 393     runner_cfg = scenario_cfg["runner"]
 394     runner_cfg['output_filename'] = output_file
 395
 396     # TODO support get multi hosts/vms info
 397     context_cfg = {}
 398     if "host" in scenario_cfg:
 399         context_cfg['host'] = Context.get_server(scenario_cfg["host"])
 400
 401     if "target" in scenario_cfg:
 402         if is_ip_addr(scenario_cfg["target"]):
 403             context_cfg['target'] = {}
 404             context_cfg['target']["ipaddr"] = scenario_cfg["target"]
 405         else:
 406             context_cfg['target'] = Context.get_server(scenario_cfg["target"])
 407             if _is_same_heat_context(scenario_cfg["host"],
 408                                      scenario_cfg["target"]):
 409                 context_cfg["target"]["ipaddr"] = \
 410                     context_cfg["target"]["private_ip"]
 411             else:
 412                 context_cfg["target"]["ipaddr"] = \
 413                     context_cfg["target"]["ip"]
 414
 415     if "targets" in scenario_cfg:
 416         ip_list = []
 417         for target in scenario_cfg["targets"]:
 418             if is_ip_addr(target):
 419                 ip_list.append(target)
 420                 context_cfg['target'] = {}
 421             else:
 422                 context_cfg['target'] = Context.get_server(target)
 423                 if _is_same_heat_context(scenario_cfg["host"], target):
 424                     ip_list.append(context_cfg["target"]["private_ip"])
 425                 else:
 426                     ip_list.append(context_cfg["target"]["ip"])
 427         context_cfg['target']['ipaddr'] = ','.join(ip_list)
 428
 429     if "nodes" in scenario_cfg:
 430         context_cfg["nodes"] = parse_nodes_with_context(scenario_cfg)
 431     runner = base_runner.Runner.get(runner_cfg)
 432
 433     print "Starting runner of type '%s'" % runner_cfg["type"]
 434     runner.run(scenario_cfg, context_cfg)
 435
 436     return runner
 437
 438
 439 def parse_nodes_with_context(scenario_cfg):
 440     '''paras the 'nodes' fields in scenario '''
 441     nodes = scenario_cfg["nodes"]
 442
 443     nodes_cfg = {}
 444     for nodename in nodes:
 445         nodes_cfg[nodename] = Context.get_server(nodes[nodename])
 446
 447     return nodes_cfg
 448
 449
 450 def runner_join(runner):
 451     '''join (wait for) a runner, exit process at runner failure'''
 452     status = runner.join()
 453     base_runner.Runner.release(runner)
 454     if status != 0:
 455         sys.exit("Runner failed")
 456
 457
 458 def print_invalid_header(source_name, args):
 459     print(("Invalid %(source)s passed:\n\n %(args)s\n")
 460           % {"source": source_name, "args": args})
 461
 462
 463 def parse_task_args(src_name, args):
 464     try:
 465         kw = args and yaml.safe_load(args)
 466         kw = {} if kw is None else kw
 467     except yaml.parser.ParserError as e:
 468         print_invalid_header(src_name, args)
 469         print(("%(source)s has to be YAML. Details:\n\n%(err)s\n")
 470               % {"source": src_name, "err": e})
 471         raise TypeError()
 472
 473     if not isinstance(kw, dict):
 474         print_invalid_header(src_name, args)
 475         print(("%(src)s had to be dict, actually %(src_type)s\n")
 476               % {"src": src_name, "src_type": type(kw)})
 477         raise TypeError()
 478     return kw