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