Expose API for cluster settings

With this commit we introduce a new `put-settings` operation that can be
used to update cluster settings via the REST API. We also deprecate the
track property `cluster-settings` which had a similar purpose but the
cluster settings ended up in `elasticsearch.yml` instead of being
updated via an API. This is now tricky as we will move away from an
integrated cluster management (see also #830) and we should instead add
settings that need to be persistent in `elasticsearch.yml` via
`--car-params` and settings that are per track via the cluster settings
API.

Relates elastic/rally-tracks#93
Relates #831

docs/migrate.rst

@@ -4,6 +4,11 @@ Migration Guide
 Migrating to Rally 1.4.0
 ------------------------
 
+cluster-settings is deprecated in favor of the put-settings operation
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Before Rally 1.4.0, cluster settings could be specified on the track with the ``cluster-settings`` property. This functionality is deprecated and you should set dynamic cluster settings via the new ``put-settings`` runner. Static settings should instead be set via ``--car-params``.
+
 Build logs are stored in Rally's log directory
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

docs/track.rst

@@ -435,6 +435,29 @@ This example requires that the ``ingest-geoip`` Elasticsearch plugin is installe
 
 This is an administrative operation. Metrics are not reported by default. Reporting can be forced by setting ``include-in-reporting`` to ``true``.
 
+put-settings
+~~~~~~~~~~~~
+
+With the operation-type ``put-settings`` you can execute the `cluster update settings API <http://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-update-settings.html>`_. It supports the following properties:
+
+* `body` (mandatory): The cluster settings to apply.
+
+Example::
+
+    {
+      "name": "increase-watermarks",
+      "operation-type": "put-settings",
+      "body": {
+        "transient" : {
+            "cluster.routing.allocation.disk.watermark.low" : "95%",
+            "cluster.routing.allocation.disk.watermark.high" : "97%",
+            "cluster.routing.allocation.disk.watermark.flood_stage" : "99%"
+        }
+      }
+    }
+
+This is an administrative operation. Metrics are not reported by default. Reporting can be forced by setting ``include-in-reporting`` to ``true``.
+
 cluster-health
 ~~~~~~~~~~~~~~
 

esrally/driver/runner.py

@@ -60,6 +60,7 @@ def register_default_runners():
     register_runner(track.OperationType.DeleteSnapshotRepository.name, Retry(DeleteSnapshotRepository()))
     register_runner(track.OperationType.CreateSnapshotRepository.name, Retry(CreateSnapshotRepository()))
     register_runner(track.OperationType.WaitForRecovery.name, Retry(IndicesRecovery()))
+    register_runner(track.OperationType.PutSettings.name, Retry(PutSettings()))
 
 
 def runner_for(operation_type):
@@ -1365,6 +1366,18 @@ def __repr__(self, *args, **kwargs):
         return "wait-for-recovery"
 
 
+class PutSettings(Runner):
+    """
+    Updates cluster settings with the
+    `cluster settings API <http://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-update-settings.html>_.
+    """
+    def __call__(self, es, params):
+        es.cluster.put_settings(body=mandatory(params, "body", repr(self)))
+
+    def __repr__(self, *args, **kwargs):
+        return "put-settings"
+
+
 # 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, Delegator):

esrally/track/loader.py

@@ -1058,6 +1058,9 @@ def _create_challenges(self, track_spec):
             # 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)
             cluster_settings = self._r(challenge_spec, "cluster-settings", error_ctx=name, mandatory=False)
+            if cluster_settings:
+                console.warn("Track [{}] uses the deprecated property [cluster-settings]. Please replace it with an "
+                             "explicit call to the cluster settings API.".format(self.name), logger=self.logger)
 
             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."

esrally/track/track.py

@@ -441,6 +441,7 @@ class OperationType(Enum):
     CreateSnapshotRepository = 1020
     RestoreSnapshot = 1021
     WaitForRecovery = 1022
+    PutSettings = 1023
 
     @property
     def admin_op(self):
@@ -502,6 +503,8 @@ def from_hyphenated_string(cls, v):
             return OperationType.RestoreSnapshot
         elif v == "wait-for-recovery":
             return OperationType.WaitForRecovery
+        elif v == "put-settings":
+            return OperationType.PutSettings
         else:
             raise KeyError("No enum value for [%s]" % v)
 

tests/driver/runner_test.py

@@ -2477,6 +2477,27 @@ def test_shrink_index_derives_shrink_node(self, sleep, es):
         })
 
 
+class PutSettingsTests(TestCase):
+    @mock.patch("elasticsearch.Elasticsearch")
+    def test_put_settings(self, es):
+        params = {
+            "body": {
+                "transient": {
+                    "indices.recovery.max_bytes_per_sec": "20mb"
+                }
+            }
+        }
+
+        r = runner.PutSettings()
+        r(es, params)
+
+        es.cluster.put_settings.assert_called_once_with(body={
+            "transient": {
+                "indices.recovery.max_bytes_per_sec": "20mb"
+            }
+        })
+
+
 class RetryTests(TestCase):
     def test_is_transparent_on_success_when_no_retries(self):
         delegate = mock.Mock()