Repository: elastic/rally

Issue:
Align operation parameter names with runner implementation
For some operations, e.g. `search`, internal parameter names (e.g. `cache`), differ from the internal name (`use_request_cache`) which is confusing to advanced users that look at our internal runners.

Hence, we should align them to avoid this.

Discussion:

Patch:
diff --git a/docs/adding_tracks.rst b/docs/adding_tracks.rst
index f8bf7bad..5b0b43f5 100644
--- a/docs/adding_tracks.rst
+++ b/docs/adding_tracks.rst
@@ -578,10 +578,6 @@ The data set that is used in the http_logs track starts on 26-04-1998 but we wan
 Custom parameter sources
 ^^^^^^^^^^^^^^^^^^^^^^^^
 
-.. note::
-
-    This is a rather new feature and the API may change! However, the effort to use custom parameter sources is very low.
-
 .. warning::
 
     Your parameter source is on a performance-critical code-path so please double-check with :ref:`Rally's profiling support <clr_enable_driver_profiling>` that you did not introduce any bottlenecks.
@@ -644,7 +640,7 @@ Rally will recognize the parameter source and looks then for a file ``track.py``
             },
             "index": index_name,
             "type": type_name,
-            "use_request_cache": params.get("cache", False)
+            "cache": params.get("cache", False)
         }
 
     def register(registry):
@@ -712,7 +708,7 @@ If you need more control, you need to implement a class. The example above, impl
                 },
                 "index": self._index_name,
                 "type": self._type_name,
-                "use_request_cache": self._cache
+                "cache": self._cache
             }
 
 
diff --git a/docs/migrate.rst b/docs/migrate.rst
index 9050cfb4..067e1582 100644
--- a/docs/migrate.rst
+++ b/docs/migrate.rst
@@ -69,6 +69,21 @@ The example above also shows how to provide per-challenge index settings. If per
 
 This behavior applies similarly to index templates as well.
 
+Custom Parameter Sources
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+We have aligned the internal names between parameter sources and runners with the ones that are specified by the user in the track file. If you have implemented custom parameter sources or runners, please adjust the parameter names as follows:
+
+============== ======================= =======================
+Operation type Old name                New name
+============== ======================= =======================
+search         use_request_cache       cache
+search         request_params          request-params
+search         items_per_page          results-per-page
+bulk           action_metadata_present action-metadata-present
+force-merge    max_num_segments        max-num-segments
+============== ======================= =======================
+
 Migrating to Rally 0.9.0
 ------------------------
 
diff --git a/docs/track.rst b/docs/track.rst
index 0bea3c76..669955df 100644
--- a/docs/track.rst
+++ b/docs/track.rst
@@ -330,7 +330,7 @@ force-merge
 
 With the operation type ``force-merge`` you can call the `force merge API <http://www.elastic.co/guide/en/elasticsearch/reference/current/indices-forcemerge.html>`_. On older versions of Elasticsearch (prior to 2.1), Rally will use the ``optimize API`` instead. It supports the following parameter:
 
-* ``max_num_segments`` (optional)  The number of segments the index should be merged into. Defaults to simply checking if a merge needs to execute, and if so, executes it.
+* ``max-num-segments`` (optional)  The number of segments the index should be merged into. Defaults to simply checking if a merge needs to execute, and if so, executes it.
 
 This is an administrative operation. Metrics are not reported by default. If reporting is forced by setting ``include-in-reporting`` to ``true``, then throughput is reported as the number of completed force-merge operations per second.
 
diff --git a/esrally/driver/runner.py b/esrally/driver/runner.py
index 4ba93457..42053c0a 100644
--- a/esrally/driver/runner.py
+++ b/esrally/driver/runner.py
@@ -254,7 +254,15 @@ class BulkIndex(Runner):
         if "pipeline" in params:
             bulk_params["pipeline"] = params["pipeline"]
 
-        with_action_metadata = mandatory(params, "action_metadata_present", self)
+        # TODO: Remove this fallback logic with Rally 1.0
+        if "action_metadata_present" in params:
+            logger.warning("Your parameter source uses the deprecated name [action_metadata_present]. Please change it to "
+                           "[action-metadata-present].")
+            action_meta_data_key = "action_metadata_present"
+        else:
+            action_meta_data_key = "action-metadata-present"
+
+        with_action_metadata = mandatory(params, action_meta_data_key, self)
         bulk_size = mandatory(params, "bulk-size", self)
 
         if with_action_metadata:
@@ -287,7 +295,15 @@ class BulkIndex(Runner):
         for line_number, data in enumerate(params["body"]):
 
             line_size = len(data.encode('utf-8'))
-            if params["action_metadata_present"]:
+
+            # TODO: Remove this fallback logic with Rally 1.0
+            if "action_metadata_present" in params:
+                logger.warning("Your parameter source uses the deprecated name [action_metadata_present]. Please change it to "
+                               "[action-metadata-present].")
+                action_meta_data_key = "action_metadata_present"
+            else:
+                action_meta_data_key = "action-metadata-present"
+            if params[action_meta_data_key]:
                 if line_number % 2 == 1:
                     total_document_size_bytes += line_size
             else:
@@ -377,15 +393,23 @@ class ForceMerge(Runner):
         logger.info("Force merging all indices.")
         import elasticsearch
         try:
-            if "max_num_segments" in params:
-                es.indices.forcemerge(index="_all", max_num_segments=params["max_num_segments"])
+            if "max-num-segments" in params:
+                max_num_segments = params["max-num-segments"]
+            elif "max_num_segments" in params:
+                logger.warning("Your parameter source uses the deprecated name [max_num_segments]. Please change it to [max-num-segments].")
+                max_num_segments = params["max_num_segments"]
+            else:
+                max_num_segments = None
+
+            if max_num_segments:
+                es.indices.forcemerge(index="_all", max_num_segments=max_num_segments)
             else:
                 es.indices.forcemerge(index="_all")
         except elasticsearch.TransportError as e:
             # this is caused by older versions of Elasticsearch (< 2.1), fall back to optimize
             if e.status_code == 400:
-                if "max_num_segments" in params:
-                    es.transport.perform_request("POST", "/_optimize?max_num_segments=%s" % (params["max_num_segments"]))
+                if max_num_segments:
+                    es.transport.perform_request("POST", "/_optimize?max_num_segments={}".format(max_num_segments))
                 else:
                     es.transport.perform_request("POST", "/_optimize")
             else:
@@ -427,14 +451,14 @@ class Query(Runner):
 
     * `index`: The index or indices against which to issue the query.
     * `type`: See `index`
-    * `use_request_cache`: True iff the request cache should be used.
+    * `cache`: True iff the request cache should be used.
     * `body`: Query body
 
     If the following parameters are present in addition, a scroll query will be issued:
 
     * `pages`: Number of pages to retrieve at most for this scroll. If a scroll query does yield less results than the specified number of
                pages we will terminate earlier.
-    * `items_per_page`: Number of items to retrieve per page.
+    * `results-per-page`: Number of results to retrieve per page.
 
     Returned meta data
 
@@ -457,14 +481,26 @@ class Query(Runner):
         self.es = None
 
     def __call__(self, es, params):
-        if "pages" in params and "items_per_page" in params:
+        # TODO: Remove items_per_page with Rally 1.0.
+        if "pages" in params and ("results-per-page" in params or "items_per_page" in params):
             return self.scroll_query(es, params)
         else:
             return self.request_body_query(es, params)
 
     def request_body_query(self, es, params):
-        request_params = params.get("request_params", {})
-        if "use_request_cache" in params:
+        if "request-params" in params:
+            request_params = params["request-params"]
+        elif "request_params" in params:
+            # TODO: Remove with Rally 1.0.
+            logger.warning("Your parameter source uses the deprecated name [request_params]. Please change it to [request-params].")
+            request_params = params["request_params"]
+        else:
+            request_params = {}
+        if "cache" in params:
+            request_params["request_cache"] = params["cache"]
+        elif "use_request_cache" in params:
+            # TODO: Remove with Rally 1.0.
+            logger.warning("Your parameter source uses the deprecated name [use_request_cache]. Please change it to [cache].")
             request_params["request_cache"] = params["use_request_cache"]
         r = es.search(
             index=params.get("index", "_all"),
@@ -481,7 +517,14 @@ class Query(Runner):
         }
 
     def scroll_query(self, es, params):
-        request_params = params.get("request_params", {})
+        if "request-params" in params:
+            request_params = params["request-params"]
+        elif "request_params" in params:
+            # TODO: Remove with Rally 1.0.
+            logger.warning("Your parameter source uses the deprecated name [request_params]. Please change it to [request-params].")
+            request_params = params["request_params"]
+        else:
+            request_params = {}
         hits = 0
         retrieved_pages = 0
         timed_out = False
@@ -489,6 +532,22 @@ class Query(Runner):
         self.es = es
         # explicitly convert to int to provoke an error otherwise
         total_pages = sys.maxsize if params["pages"] == "all" else int(params["pages"])
+        if "cache" in params:
+            cache = params["cache"]
+        elif "use_request_cache" in params:
+            # TODO: Remove with Rally 1.0.
+            logger.warning("Your parameter source uses the deprecated name [use_request_cache]. Please change it to [cache].")
+            cache = params["use_request_cache"]
+        else:
+            cache = None
+        if "results-per-page" in params:
+            size = params["results-per-page"]
+        elif "items_per_page" in params:
+            # TODO: Remove with Rally 1.0.
+            logger.warning("Your parameter source uses the deprecated name [items_per_page]. Please change it to [results-per-page].")
+            size = params["items_per_page"]
+        else:
+            size = None
 
         for page in range(total_pages):
             if page == 0:
@@ -498,8 +557,8 @@ class Query(Runner):
                     body=mandatory(params, "body", self),
                     sort="_doc",
                     scroll="10s",
-                    size=params["items_per_page"],
-                    request_cache=params.get("use_request_cache"),
+                    size=size,
+                    request_cache=cache,
                     **request_params
                 )
                 # This should only happen if we concurrently create an index and start searching
@@ -615,7 +674,7 @@ class PutPipeline(Runner):
     def __call__(self, es, params):
         es.ingest.put_pipeline(id=mandatory(params, "id", self),
                                body=mandatory(params, "body", self),
-                               master_timeout=params.get("master_timeout"),
+                               master_timeout=params.get("master-timeout"),
                                timeout=params.get("timeout"),
                                )
 
diff --git a/esrally/track/params.py b/esrally/track/params.py
index 70cd3ace..28683fe0 100644
--- a/esrally/track/params.py
+++ b/esrally/track/params.py
@@ -353,13 +353,17 @@ class SearchParamSource(ParamSource):
         query_body = params.get("body", None)
         query_body_params = params.get("body-params", None)
         pages = params.get("pages", None)
-        items_per_page = params.get("results-per-page", None)
+        results_per_page = params.get("results-per-page", None)
         request_params = params.get("request-params", {})
 
         self.query_params = {
             "index": index_name,
             "type": type_name,
+            "cache": request_cache,
+            # TODO: This is the old name, remove with Rally 1.0
             "use_request_cache": request_cache,
+            "request-params": request_params,
+            # TODO: This is the old name, remove with Rally 1.0
             "request_params": request_params,
             "body": query_body
         }
@@ -369,13 +373,15 @@ class SearchParamSource(ParamSource):
 
         if pages:
             self.query_params["pages"] = pages
-        if items_per_page:
-            self.query_params["items_per_page"] = items_per_page
+        if results_per_page:
+            self.query_params["results-per-page"] = results_per_page
+            # TODO: This is the old name, remove with Rally 1.0
+            self.query_params["items_per_page"] = results_per_page
 
         self.query_body_params = []
         if query_body_params:
             for param, data in query_body_params.items():
-                # TODO #365: Stricly check for allowed syntax. Be lenient in the pre-release and only interpret what's safely possible.
+                # 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 "
@@ -655,6 +661,8 @@ def bulk_generator(readers, client_index, pipeline, original_params):
                 "type": type,
                 # For our implementation it's always present. Either the original source file already contains this line or the generator
                 # has added it.
+                "action-metadata-present": True,
+                # TODO: This is the old name, remove with Rally 1.0
                 "action_metadata_present": True,
                 "body": bulk,
                 # This is not always equal to the bulk_size we get as parameter. The last bulk may be less than the bulk size.


Test patch:
diff --git a/tests/driver/runner_test.py b/tests/driver/runner_test.py
index 8ad55c61..dc7462ae 100644
--- a/tests/driver/runner_test.py
+++ b/tests/driver/runner_test.py
@@ -579,7 +579,7 @@ class QueryRunnerTests(TestCase):
         params = {
             "index": "unittest",
             "type": "type",
-            "use_request_cache": False,
+            "cache": False,
             "body": {
                 "query": {
                     "match_all": {}
@@ -626,10 +626,10 @@ class QueryRunnerTests(TestCase):
 
         params = {
             "pages": 1,
-            "items_per_page": 100,
+            "results-per-page": 100,
             "index": "unittest",
             "type": "type",
-            "use_request_cache": False,
+            "cache": False,
             "body": {
                 "query": {
                     "match_all": {}
@@ -677,7 +677,7 @@ class QueryRunnerTests(TestCase):
 
         params = {
             "pages": 1,
-            "items_per_page": 100,
+            "results-per-page": 100,
             "body": {
                 "query": {
                     "match_all": {}
@@ -738,10 +738,10 @@ class QueryRunnerTests(TestCase):
 
         params = {
             "pages": 2,
-            "items_per_page": 100,
+            "results-per-page": 100,
             "index": "unittest",
             "type": "type",
-            "use_request_cache": False,
+            "cache": False,
             "body": {
                 "query": {
                     "match_all": {}
@@ -795,10 +795,10 @@ class QueryRunnerTests(TestCase):
 
         params = {
             "pages": 5,
-            "items_per_page": 100,
+            "results-per-page": 100,
             "index": "unittest",
             "type": "type",
-            "use_request_cache": False,
+            "cache": False,
             "body": {
                 "query": {
                     "match_all": {}
@@ -860,10 +860,10 @@ class QueryRunnerTests(TestCase):
 
         params = {
             "pages": "all",
-            "items_per_page": 100,
+            "results-per-page": 100,
             "index": "unittest",
             "type": "type",
-            "use_request_cache": False,
+            "cache": False,
             "body": {
                 "query": {
                     "match_all": {}
@@ -922,7 +922,7 @@ class PutPipelineRunnerTests(TestCase):
         es.ingest.put_pipeline.assert_not_called()
 
     @mock.patch("elasticsearch.Elasticsearch")
-    def test_param_body_mandatory(self, es):
+    def test_param_id_mandatory(self, es):
         r = runner.PutPipeline()
 
         params = {
diff --git a/tests/track/params_test.py b/tests/track/params_test.py
index f4fa5498..eed38846 100644
--- a/tests/track/params_test.py
+++ b/tests/track/params_test.py
@@ -699,6 +699,7 @@ class BulkDataGeneratorTests(TestCase):
         all_bulks = list(bulks)
         self.assertEqual(2, len(all_bulks))
         self.assertEqual({
+            "action-metadata-present": True,
             "action_metadata_present": True,
             "body": ["1", "2", "3", "4", "5"],
             "bulk-id": "0-1",
@@ -710,6 +711,7 @@ class BulkDataGeneratorTests(TestCase):
         }, all_bulks[0])
 
         self.assertEqual({
+            "action-metadata-present": True,
             "action_metadata_present": True,
             "body": ["6", "7", "8"],
             "bulk-id": "0-2",
@@ -755,6 +757,7 @@ class BulkDataGeneratorTests(TestCase):
         all_bulks = list(bulks)
         self.assertEqual(3, len(all_bulks))
         self.assertEqual({
+            "action-metadata-present": True,
             "action_metadata_present": True,
             "body": ["1", "2", "3", "4", "5"],
             "bulk-id": "0-1",
@@ -766,6 +769,7 @@ class BulkDataGeneratorTests(TestCase):
         }, all_bulks[0])
 
         self.assertEqual({
+            "action-metadata-present": True,
             "action_metadata_present": True,
             "body": ["1", "2", "3", "4", "5"],
             "bulk-id": "0-2",
@@ -777,6 +781,7 @@ class BulkDataGeneratorTests(TestCase):
         }, all_bulks[1])
 
         self.assertEqual({
+            "action-metadata-present": True,
             "action_metadata_present": True,
             "body": ["1", "2", "3", "4", "5"],
             "bulk-id": "0-3",
@@ -807,6 +812,7 @@ class BulkDataGeneratorTests(TestCase):
         self.assertEqual(1, len(all_bulks))
         # body must not contain 'foo'!
         self.assertEqual({
+            "action-metadata-present": True,
             "action_metadata_present": True,
             "body": ["1", "2", "3"],
             "bulk-id": "0-1",
@@ -1265,18 +1271,24 @@ class SearchParamSourceTests(TestCase):
         })
         p = source.params()
 
-        self.assertEqual(5, len(p))
+        self.assertEqual(7, len(p))
         self.assertEqual("index1", p["index"])
         self.assertEqual("type1", p["type"])
         self.assertEqual({
             "_source_include": "some_field"
-        }, p["request_params"])
-        self.assertFalse(p["use_request_cache"])
+        }, p["request-params"])
+        self.assertFalse(p["cache"])
         self.assertEqual({
             "query": {
                 "match_all": {}
             }
         }, p["body"])
+        # backwards-compatibility options
+        self.assertFalse(p["use_request_cache"])
+        self.assertEqual({
+            "_source_include": "some_field"
+        }, p["request_params"])
+
 
     def test_replaces_body_params(self):
         import copy