Repository: elastic/rally

Issue:
Add configurable "sleep" operation
When running benchmarks against nodes that are expected to see few but long-running queries it is possible to combine request timeout with `target-interval` to ensure there is a bit of quiet time between queries. This however results in a longer quiet time for queries that return quickly. It would be useful to have an operation that just adds a configurable pause to the track so that the same quiet time can be achieved no matter how fast the query returns.

Discussion:

Patch:
diff --git a/docs/command_line_reference.rst b/docs/command_line_reference.rst
index 42858950..6146b16d 100644
--- a/docs/command_line_reference.rst
+++ b/docs/command_line_reference.rst
@@ -64,6 +64,11 @@ Examples::
 
 Selects the track repository that Rally should use to resolve tracks. By default the ``default`` track repository is used, which is available in the Github project `rally-tracks <https://github.com/elastic/rally-tracks>`_. See the :doc:`track reference </track>` on how to add your own track repositories. ``--track-path`` and ``--track-repository`` as well as ``--track`` are mutually exclusive.
 
+``track-revision``
+~~~~~~~~~~~~~~~~~~
+
+Selects a specific revision in the track repository. By default, Rally will choose the most appropriate branch on its own but in some cases it is necessary to specify a certain commit. This is mostly needed when testing whether a change in performance has occurred due to a change in the workload. 
+
 ``track``
 ~~~~~~~~~
 
diff --git a/docs/track.rst b/docs/track.rst
index f477b563..bf632cce 100644
--- a/docs/track.rst
+++ b/docs/track.rst
@@ -793,6 +793,18 @@ With the operation ``raw-request`` you can execute arbitrary HTTP requests again
 * ``request-params`` (optional): A structure containing HTTP request parameters.
 * ``ignore`` (optional): An array of HTTP response status codes to ignore (i.e. consider as successful).
 
+sleep
+~~~~~
+
+With the operation ``sleep`` you can sleep for a certain duration to ensure no requests are executed by the corresponding clients. The ``sleep`` operation supports the following parameter:
+
+* ``duration`` (mandatory): A non-negative number that defines the sleep duration in seconds.
+
+.. note::
+    The ``sleep`` operation is only useful in very limited circumstances. To throttle throughput, specify a ``target-throughput`` on the corresponding task instead.
+
+This is an administrative operation. Metrics are not reported by default. Reporting can be forced by setting ``include-in-reporting`` to ``true``.
+
 schedule
 ~~~~~~~~
 
diff --git a/esrally/driver/runner.py b/esrally/driver/runner.py
index 1b13228c..5223f96d 100644
--- a/esrally/driver/runner.py
+++ b/esrally/driver/runner.py
@@ -36,6 +36,8 @@ def register_default_runners():
     register_runner(track.OperationType.NodesStats.name, NodeStats())
     register_runner(track.OperationType.Search.name, Query())
     register_runner(track.OperationType.RawRequest.name, RawRequest())
+    # This is an administrative operation but there is no need for a retry here as we don't issue a request
+    register_runner(track.OperationType.Sleep.name, Sleep())
 
     # We treat the following as administrative commands and thus already start to wrap them in a retry.
     register_runner(track.OperationType.ClusterHealth.name, Retry(ClusterHealth()))
@@ -1070,6 +1072,18 @@ class RawRequest(Runner):
         return "raw-request"
 
 
+class Sleep(Runner):
+    """
+    Sleeps for the specified duration not issuing any request.
+    """
+
+    def __call__(self, es, params):
+        time.sleep(mandatory(params, "duration", "sleep"))
+
+    def __repr__(self, *args, **kwargs):
+        return "sleep"
+
+
 # TODO: Allow to use this from (selected) regular runners and add user documentation.
 # TODO: It would maybe be interesting to add meta-data on how many retries there were.
 class Retry(Runner):
diff --git a/esrally/rally.py b/esrally/rally.py
index 3a89ec35..b4e91454 100644
--- a/esrally/rally.py
+++ b/esrally/rally.py
@@ -167,6 +167,10 @@ def create_arg_parser():
         track_source_group.add_argument(
             "--track-path",
             help="Define the path to a track.")
+        track_source_group.add_argument(
+            "--track-revision",
+            help="Define a specific revision in the track repository that Rally should use.",
+            default=None)
         p.add_argument(
             "--team-repository",
             help="Define the repository from where Rally will load teams and cars (default: default).",
@@ -573,6 +577,7 @@ def main():
     if args.track_path:
         cfg.add(config.Scope.applicationOverride, "track", "track.path", os.path.abspath(io.normalize_path(args.track_path)))
         cfg.add(config.Scope.applicationOverride, "track", "repository.name", None)
+        cfg.add(config.Scope.applicationOverride, "track", "repository.revision", None)
         if args.track:
             # stay as close as possible to argparse errors although we have a custom validation.
             arg_parser.error("argument --track not allowed with argument --track-path")
@@ -580,6 +585,7 @@ def main():
     else:
         # cfg.add(config.Scope.applicationOverride, "track", "track.path", None)
         cfg.add(config.Scope.applicationOverride, "track", "repository.name", args.track_repository)
+        cfg.add(config.Scope.applicationOverride, "track", "repository.revision", args.track_revision)
         # set the default programmatically because we need to determine whether the user has provided a value
         chosen_track = args.track if args.track else "geonames"
         cfg.add(config.Scope.applicationOverride, "track", "track.name", chosen_track)
diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index 8035a12e..ebccef34 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -185,6 +185,7 @@ class GitTrackRepository:
         self.track_name = cfg.opts("track", "track.name", mandatory=False)
         distribution_version = cfg.opts("mechanic", "distribution.version", mandatory=False)
         repo_name = cfg.opts("track", "repository.name")
+        repo_revision = cfg.opts("track", "repository.revision", mandatory=False)
         offline = cfg.opts("system", "offline.mode")
         remote_url = cfg.opts("tracks", "%s.url" % repo_name, mandatory=False)
         root = cfg.opts("node", "root.dir")
@@ -193,7 +194,10 @@ class GitTrackRepository:
 
         self.repo = repo_class(remote_url, tracks_dir, repo_name, "tracks", offline, fetch)
         if update:
-            self.repo.update(distribution_version)
+            if repo_revision:
+                self.repo.checkout(repo_revision)
+            else:
+                self.repo.update(distribution_version)
 
     @property
     def track_names(self):
diff --git a/esrally/track/params.py b/esrally/track/params.py
index 6b342f43..a87bca49 100644
--- a/esrally/track/params.py
+++ b/esrally/track/params.py
@@ -21,6 +21,7 @@ import time
 import math
 import types
 import operator
+import numbers
 from enum import Enum
 
 from esrally import exceptions
@@ -130,6 +131,23 @@ class DelegatingParamSource(ParamSource):
         return self.delegate(self.track, self._params, **self.kwargs)
 
 
+class SleepParamSource(ParamSource):
+    def __init__(self, track, params, **kwargs):
+        super().__init__(track, params, **kwargs)
+        try:
+            duration = params["duration"]
+        except KeyError:
+            raise exceptions.InvalidSyntax("parameter 'duration' is mandatory for sleep operation")
+
+        if not isinstance(duration, numbers.Number):
+            raise exceptions.InvalidSyntax("parameter 'duration' for sleep operation must be a number")
+        if duration < 0:
+            raise exceptions.InvalidSyntax("parameter 'duration' must be non-negative but was {}".format(duration))
+
+    def params(self):
+        return dict(self._params)
+
+
 class CreateIndexParamSource(ParamSource):
     def __init__(self, track, params, **kwargs):
         super().__init__(track, params, **kwargs)
@@ -889,6 +907,7 @@ register_param_source_for_operation(track.OperationType.CreateIndex, CreateIndex
 register_param_source_for_operation(track.OperationType.DeleteIndex, DeleteIndexParamSource)
 register_param_source_for_operation(track.OperationType.CreateIndexTemplate, CreateIndexTemplateParamSource)
 register_param_source_for_operation(track.OperationType.DeleteIndexTemplate, DeleteIndexTemplateParamSource)
+register_param_source_for_operation(track.OperationType.Sleep, SleepParamSource)
 
 # Also register by name, so users can use it too
 register_param_source_for_name("file-reader", BulkIndexParamSource)
diff --git a/esrally/track/track.py b/esrally/track/track.py
index e3e382d6..0dde287f 100644
--- a/esrally/track/track.py
+++ b/esrally/track/track.py
@@ -436,6 +436,7 @@ class OperationType(Enum):
     DeleteMlJob = 1015
     OpenMlJob = 1016
     CloseMlJob = 1017
+    Sleep = 1018
 
     @property
     def admin_op(self):
@@ -487,6 +488,8 @@ class OperationType(Enum):
             return OperationType.OpenMlJob
         elif v == "close-ml-job":
             return OperationType.CloseMlJob
+        elif v == "sleep":
+            return OperationType.Sleep
         else:
             raise KeyError("No enum value for [%s]" % v)
 


Test patch:
diff --git a/tests/driver/runner_test.py b/tests/driver/runner_test.py
index fe344e38..2ef53314 100644
--- a/tests/driver/runner_test.py
+++ b/tests/driver/runner_test.py
@@ -1798,6 +1798,31 @@ class RawRequestRunnerTests(TestCase):
                                                              params={})
 
 
+class SleepTests(TestCase):
+    @mock.patch("elasticsearch.Elasticsearch")
+    # To avoid real sleeps in unit tests
+    @mock.patch("time.sleep")
+    def test_missing_parameter(self, sleep, es):
+        r = runner.Sleep()
+        with self.assertRaisesRegex(exceptions.DataError,
+                                    "Parameter source for operation 'sleep' did not provide the mandatory parameter "
+                                    "'duration'. Please add it to your parameter source."):
+            r(es, params={})
+
+        self.assertEqual(0, es.call_count)
+        self.assertEqual(0, sleep.call_count)
+
+    @mock.patch("elasticsearch.Elasticsearch")
+    # To avoid real sleeps in unit tests
+    @mock.patch("time.sleep")
+    def test_sleep(self, sleep, es):
+        r = runner.Sleep()
+        r(es, params={"duration": 4.3})
+
+        self.assertEqual(0, es.call_count)
+        sleep.assert_called_once_with(4.3)
+
+
 class ShrinkIndexTests(TestCase):
     @mock.patch("elasticsearch.Elasticsearch")
     # To avoid real sleeps in unit tests
diff --git a/tests/track/params_test.py b/tests/track/params_test.py
index b94d9c72..8c3b7dda 100644
--- a/tests/track/params_test.py
+++ b/tests/track/params_test.py
@@ -1184,6 +1184,26 @@ class ParamsRegistrationTests(TestCase):
         params._unregister_param_source_for_name(source_name)
 
 
+class SleepParamSourceTests(TestCase):
+    def test_missing_duration_parameter(self):
+        with self.assertRaisesRegex(exceptions.InvalidSyntax, "parameter 'duration' is mandatory for sleep operation"):
+            params.SleepParamSource(track.Track(name="unit-test"), params={})
+
+    def test_duration_parameter_wrong_type(self):
+        with self.assertRaisesRegex(exceptions.InvalidSyntax,
+                                    "parameter 'duration' for sleep operation must be a number"):
+            params.SleepParamSource(track.Track(name="unit-test"), params={"duration": "this is a string"})
+
+    def test_duration_parameter_negative_number(self):
+        with self.assertRaisesRegex(exceptions.InvalidSyntax,
+                                    "parameter 'duration' must be non-negative but was -1.0"):
+            params.SleepParamSource(track.Track(name="unit-test"), params={"duration": -1.0})
+
+    def test_param_source_passes_all_parameters(self):
+        p = params.SleepParamSource(track.Track(name="unit-test"), params={"duration": 3.4, "additional": True})
+        self.assertDictEqual({"duration": 3.4, "additional": True}, p.params())
+
+
 class CreateIndexParamSourceTests(TestCase):
     def test_create_index_inline_with_body(self):
         source = params.CreateIndexParamSource(track.Track(name="unit-test"), params={