How to use the esrally.exceptions.RallyError function in esrally

self.query_body_params = []
        if query_body_params:
            for param, data in query_body_params.items():
                # TODO #365: Strictly check for allowed syntax. Be lenient in the pre-release and only interpret what's safely possible.
                # build path based on param
                # if not isinstance(data, list):
                #    raise exceptions.RallyError("%s in body-params defines %s but only lists are allowed. This may be a new syntax "
                #                                "that is not recognized by this version. Please upgrade Rally." % (param, data))
                if isinstance(data, list):
                    query_body_path = param.split(".")
                    b = self.query_params["body"]
                    # check early to ensure this path is actually contained in the body
                    try:
                        self.get_from_dict(b, query_body_path)
                    except KeyError:
                        raise exceptions.RallyError("The path %s could not be found within the query body %s." % (param, b))

                    self.query_body_params.append((query_body_path, data))

console.warn("This command is deprecated and will be removed with the next release of Rally.", overline="!", underline="!")
    # provide a custom error message
    target_hosts = cfg.opts("facts", "hosts", mandatory=False)
    if not target_hosts:
        raise exceptions.SystemSetupError("Please define a target host with --target-hosts")
    if len(target_hosts) > 1:
        raise exceptions.SystemSetupError("Only one target host is supported at the moment but you provided %s" % target_hosts)

    # at this point an actor system has to run and we should only join
    actor_system = actor.bootstrap_actor_system(try_join=True)
    facts_actor = actor_system.createActor(FactsActor, targetActorRequirements={"ip": target_hosts[0]})
    result = actor_system.ask(facts_actor, GatherFacts())
    if isinstance(result, Facts):
        console.println(json.dumps(result.facts, indent="  "))
    else:
        raise exceptions.RallyError("Could not gather facts: [%s]." % str(result))

version=es_version)
        suppliers.append(ElasticsearchDistributionSupplier(repo, distributions_root))

    for plugin in plugins:
        supplier_type, plugin_version, build_plugin = supply_requirements[plugin.name]

        if supplier_type == "source":
            if CorePluginSourceSupplier.can_handle(plugin):
                logger.info("Adding core plugin source supplier for [%s].", plugin.name)
                assert es_src_dir is not None, "Cannot build core plugin %s when Elasticsearch is not built from source." % plugin.name
                suppliers.append(CorePluginSourceSupplier(plugin, es_src_dir, builder))
            elif ExternalPluginSourceSupplier.can_handle(plugin):
                logger.info("Adding external plugin source supplier for [%s].", plugin.name)
                suppliers.append(ExternalPluginSourceSupplier(plugin, plugin_version, _src_dir(cfg, mandatory=False), src_config, builder))
            else:
                raise exceptions.RallyError("Plugin %s can neither be treated as core nor as external plugin. Requirements: %s" %
                                            (plugin.name, supply_requirements[plugin.name]))
        else:
            logger.info("Adding plugin distribution supplier for [%s].", plugin.name)
            assert repo is not None, "Cannot benchmark plugin %s from a distribution version but Elasticsearch from sources" % plugin.name
            suppliers.append(PluginDistributionSupplier(repo, plugin))

    return CompositeSupplier(suppliers)

def race(cfg):
    other_rally_processes = process.find_all_other_rally_processes()
    if other_rally_processes:
        pids = [p.pid for p in other_rally_processes]
        msg = "There are other Rally processes running on this machine (PIDs: %s) but only one Rally benchmark is allowed to run at " \
              "the same time. Please check and terminate these processes and retry again." % pids
        raise exceptions.RallyError(msg)

    with_actor_system(racecontrol.run, cfg)

def race(cfg, sources=False, build=False, distribution=False, external=False, docker=False):
    # at this point an actor system has to run and we should only join
    actor_system = actor.bootstrap_actor_system(try_join=True)
    benchmark_actor = actor_system.createActor(BenchmarkActor, targetActorRequirements={"coordinator": True})
    try:
        result = actor_system.ask(benchmark_actor, Setup(cfg, sources, build, distribution, external, docker))
        if isinstance(result, Success):
            logger.info("Benchmark has finished successfully.")
        # may happen if one of the load generators has detected that the user has cancelled the benchmark.
        elif isinstance(result, actor.BenchmarkCancelled):
            logger.info("User has cancelled the benchmark (detected by actor).")
        elif isinstance(result, actor.BenchmarkFailure):
            logger.error("A benchmark failure has occurred")
            raise exceptions.RallyError(result.message, result.cause)
        else:
            raise exceptions.RallyError("Got an unexpected result during benchmarking: [%s]." % str(result))
    except KeyboardInterrupt:
        logger.info("User has cancelled the benchmark (detected by race control).")
        # notify the coordinator so it can properly handle this state. Do it blocking so we don't have a race between this message
        # and the actor exit request.
        actor_system.ask(benchmark_actor, actor.BenchmarkCancelled())
    finally:
        logger.info("Telling benchmark actor to exit.")
        actor_system.tell(benchmark_actor, thespian.actors.ActorExitRequest())

logger = logging.getLogger(__name__)
    # at this point an actor system has to run and we should only join
    actor_system = actor.bootstrap_actor_system(try_join=True)
    benchmark_actor = actor_system.createActor(BenchmarkActor, targetActorRequirements={"coordinator": True})
    try:
        result = actor_system.ask(benchmark_actor, Setup(cfg, sources, build, distribution, external, docker))
        if isinstance(result, Success):
            logger.info("Benchmark has finished successfully.")
        # may happen if one of the load generators has detected that the user has cancelled the benchmark.
        elif isinstance(result, actor.BenchmarkCancelled):
            logger.info("User has cancelled the benchmark (detected by actor).")
        elif isinstance(result, actor.BenchmarkFailure):
            logger.error("A benchmark failure has occurred")
            raise exceptions.RallyError(result.message, result.cause)
        else:
            raise exceptions.RallyError("Got an unexpected result during benchmarking: [%s]." % str(result))
    except KeyboardInterrupt:
        logger.info("User has cancelled the benchmark (detected by race control).")
        # notify the coordinator so it can properly handle this state. Do it blocking so we don't have a race between this message
        # and the actor exit request.
        actor_system.ask(benchmark_actor, actor.BenchmarkCancelled())
    finally:
        logger.info("Telling benchmark actor to exit.")
        actor_system.tell(benchmark_actor, thespian.actors.ActorExitRequest())

def race(cfg, sources=False, build=False, distribution=False, external=False, docker=False):
    logger = logging.getLogger(__name__)
    # at this point an actor system has to run and we should only join
    actor_system = actor.bootstrap_actor_system(try_join=True)
    benchmark_actor = actor_system.createActor(BenchmarkActor, targetActorRequirements={"coordinator": True})
    try:
        result = actor_system.ask(benchmark_actor, Setup(cfg, sources, build, distribution, external, docker))
        if isinstance(result, Success):
            logger.info("Benchmark has finished successfully.")
        # may happen if one of the load generators has detected that the user has cancelled the benchmark.
        elif isinstance(result, actor.BenchmarkCancelled):
            logger.info("User has cancelled the benchmark (detected by actor).")
        elif isinstance(result, actor.BenchmarkFailure):
            logger.error("A benchmark failure has occurred")
            raise exceptions.RallyError(result.message, result.cause)
        else:
            raise exceptions.RallyError("Got an unexpected result during benchmarking: [%s]." % str(result))
    except KeyboardInterrupt:
        logger.info("User has cancelled the benchmark (detected by race control).")
        # notify the coordinator so it can properly handle this state. Do it blocking so we don't have a race between this message
        # and the actor exit request.
        actor_system.ask(benchmark_actor, actor.BenchmarkCancelled())
    finally:
        logger.info("Telling benchmark actor to exit.")
        actor_system.tell(benchmark_actor, thespian.actors.ActorExitRequest())

msg = "Could not connect to your Elasticsearch metrics store. Please check that it is running on host [%s] at port [%s]" \
                      " or fix the configuration in [%s]." % (node["host"], node["port"], config.ConfigFile().location)
                logger.exception(msg)
                raise exceptions.SystemSetupError(msg)
            except elasticsearch.TransportError as e:
                # gateway timeout - let's wait a bit and retry
                if e.status_code == 504 and execution_count < max_execution_count:
                    logger.info("Received a gateway timeout from the metrics store in attempt [%d/%d]." %
                                (execution_count, max_execution_count))
                    time.sleep(1)
                else:
                    node = self._client.transport.hosts[0]
                    msg = "A transport error occurred while running the operation [%s] against your Elasticsearch metrics store on " \
                          "host [%s] at port [%s]." % (target.__name__, node["host"], node["port"])
                    logger.exception(msg)
                    raise exceptions.RallyError(msg)

            except elasticsearch.exceptions.ElasticsearchException:
                node = self._client.transport.hosts[0]
                msg = "An unknown error occurred while running the operation [%s] against your Elasticsearch metrics store on host [%s] " \
                      "at port [%s]." % (target.__name__, node["host"], node["port"])
                logger.exception(msg)
                # this does not necessarily mean it's a system setup problem...
                raise exceptions.RallyError(msg)

class RallyError(Exception):
    """
    Base class for all Rally exceptions
    """

    def __init__(self, message, cause=None):
        super().__init__(message, cause)
        self.message = message
        self.cause = cause

    def __repr__(self):
        return self.message


class LaunchError(RallyError):
    """
    Thrown whenever there was a problem launching the benchmark candidate
    """


class SystemSetupError(RallyError):
    """
    Thrown when a user did something wrong, e.g. the metrics store is not started or required software is not installed
    """


class RallyAssertionError(RallyError):
    """
    Thrown when a (precondition) check has been violated.
    """