in esrally/track/loader.py [0:0]
def _create_challenges(self, track_spec):
ops = self.parse_operations(self._r(track_spec, "operations", mandatory=False, default_value=[]))
track_params = self._r(track_spec, "parameters", mandatory=False, default_value={})
challenges = []
known_challenge_names = set()
default_challenge = None
challenge_specs, auto_generated = self._get_challenge_specs(track_spec)
number_of_challenges = len(challenge_specs)
for challenge_spec in challenge_specs:
name = self._r(challenge_spec, "name", error_ctx="challenges")
description = self._r(challenge_spec, "description", error_ctx=name, mandatory=False)
user_info = self._r(challenge_spec, "user-info", error_ctx=name, mandatory=False)
challenge_params = self._r(challenge_spec, "parameters", error_ctx=name, mandatory=False, default_value={})
meta_data = self._r(challenge_spec, "meta", error_ctx=name, mandatory=False)
# if we only have one challenge it is treated as default challenge, no matter what the user has specified
default = number_of_challenges == 1 or self._r(challenge_spec, "default", error_ctx=name, mandatory=False)
selected = number_of_challenges == 1 or self.selected_challenge == name
if default and default_challenge is not None:
self._error(
"Both '%s' and '%s' are defined as default challenges. Please define only one of them as default."
% (default_challenge.name, name)
)
if name in known_challenge_names:
self._error("Duplicate challenge with name '%s'." % name)
known_challenge_names.add(name)
schedule = []
for op in self._r(challenge_spec, "schedule", error_ctx=name):
if "parallel" in op:
task = self.parse_parallel(op["parallel"], ops, name)
else:
task = self.parse_task(op, ops, name)
schedule.append(task)
# verify we don't have any duplicate task names (which can be confusing / misleading in reporting).
known_task_names = set()
for task in schedule:
for sub_task in task:
if sub_task.name in known_task_names:
self._error(
"Challenge '%s' contains multiple tasks with the name '%s'. Please use the task's name property to "
"assign a unique name for each task." % (name, sub_task.name)
)
else:
known_task_names.add(sub_task.name)
# merge params
final_challenge_params = dict(collections.merge_dicts(track_params, challenge_params))
challenge = track.Challenge(
name=name,
parameters=final_challenge_params,
meta_data=meta_data,
description=description,
user_info=user_info,
default=default,
selected=selected,
auto_generated=auto_generated,
schedule=schedule,
)
if default:
default_challenge = challenge
challenges.append(challenge)
if challenges and default_challenge is None:
self._error(
'No default challenge specified. Please edit the track and add "default": true to one of the challenges %s.'
% ", ".join([c.name for c in challenges])
)
return challenges