lint fix, add missing parameters to params.py

esrally/driver/runner.py

@@ -1559,7 +1559,8 @@ async def __call__(self, es, params):
  wait_for_completion = params.get("wait-for-completion", False)
  repository = mandatory(params, "repository", repr(self))
  snapshot = mandatory(params, "snapshot", repr(self))
- body = mandatory(params, "body", repr(self))
+ # just assert, gets set in _default_kw_params
+ mandatory(params, "body", repr(self))
  api_kwargs = self._default_kw_params(params)
  await es.snapshot.create(repository=repository,
  snapshot=snapshot,

esrally/track/params.py

@@ -127,6 +127,18 @@ def params(self):
  """
  return self._params
 
+ def _client_params(self):
+ """
+ For use when a ParamSource does not propagate self._params but does use elasticsearch client under the hood
+
+ :param params: A hash containing the source parameters from the track definition JSON
+ :return: all applicable parameters that are global to Rally and apply to the elasticsearch-py client
+ """
+ defaults = {}
+ defaults["request-timeout"] = self._params.get("request-timeout")
+ defaults["headers"] = self._params.get("headers")
+ defaults["opaque-id"] = self._params.get("opaque-id")
+ return defaults
 
 class DelegatingParamSource(ParamSource):
  def __init__(self, track, params, delegate, **kwargs):
@@ -479,6 +491,8 @@ def set_in_dict(self, d, path, val):
 
  # pylint: disable=arguments-differ
  def params(self, choice=random.choice):
+ # Ensure we pass global parameters
+ self.query_params.update(self._client_params())
  if self.query_body_params:
  # needs to replace params first
  for path, data in self.query_body_params:
@@ -700,18 +714,18 @@ def __init__(self, track, params, **kwargs):
  self._target_name = params.get("data-stream", default_target)
 
  self._max_num_segments = params.get("max-num-segments")
- self._request_timeout = params.get("request-timeout")
  self._poll_period = params.get("poll-period", 10)
  self._mode = params.get("mode", "blocking")
 
  def params(self):
- return {
+ parsed_params = {
  "index": self._target_name,
  "max-num-segments": self._max_num_segments,
- "request-timeout": self._request_timeout,
  "mode": self._mode,
  "poll-period": self._poll_period
  }
+ parsed_params.update(self._client_params())
+ return parsed_params
 
 
 def number_of_bulks(corpora, start_partition_index, end_partition_index, total_partitions, bulk_size):