Fix serverless public user in trivial tracks (#1785)

This change  eliminates the following blocking points for Rally public
use against serverless clusters:
* The `driver.Driver.prepare_benchmark()` method calls
`client.wait_for_rest_layer()` which includes `_cluster/health` ES API
call which is not available in serverless.
* The `on_benchmark_stop()` in telemetry is incorrectly trying to stop
all telemetry devices even if they were determined to be unsuitable for
serverless.
* The `DeleteIndex` runner includes a call to
`set_destructive_requires_name()` method which relies on
`_cluster/settings` API which is not available in serverless.

With the fixes in it is now possible to run trivial benchmarks with
`delete-index`, `create-index` and `bulk` operarions.

esrally/driver/driver.py

@@ -711,18 +711,18 @@ def prepare_benchmark(self, t):
 
         skip_rest_api_check = self.config.opts("mechanic", "skip.rest.api.check")
         uses_static_responses = self.config.opts("client", "options").uses_static_responses
-        serverless_mode = False
-        serverless_operator = False
+        serverless_mode = convert.to_bool(self.config.opts("driver", "serverless.mode", mandatory=False, default_value=False))
+        serverless_operator = convert.to_bool(self.config.opts("driver", "serverless.operator", mandatory=False, default_value=False))
         build_hash = None
         if skip_rest_api_check:
             self.logger.info("Skipping REST API check as requested explicitly.")
         elif uses_static_responses:
             self.logger.info("Skipping REST API check as static responses are used.")
+        elif serverless_mode and not serverless_operator:
+            self.logger.info("Skipping REST API check while targetting serverless cluster with a public user.")
         else:
             self.wait_for_rest_api(es_clients)
             self.driver_actor.cluster_details = self.retrieve_cluster_info(es_clients)
-            serverless_mode = convert.to_bool(self.config.opts("driver", "serverless.mode", mandatory=False, default_value=False))
-            serverless_operator = convert.to_bool(self.config.opts("driver", "serverless.operator", mandatory=False, default_value=False))
             if serverless_mode and serverless_operator:
                 build_hash = self.retrieve_build_hash_from_nodes_info(es_clients)
                 self.logger.info("Retrieved actual build hash [%s] from serverless cluster.", build_hash)

esrally/driver/runner.py

@@ -16,6 +16,7 @@
 # under the License.
 
 import asyncio
+import contextlib
 import contextvars
 import json
 import logging
@@ -1409,12 +1410,22 @@ class DeleteIndex(Runner):
     """
 
     async def __call__(self, es, params):
+        # pylint: disable=import-outside-toplevel
+        import elasticsearch
+
         ops = 0
 
         indices = mandatory(params, "indices", self)
         only_if_exists = params.get("only-if-exists", False)
         request_params = params.get("request-params", {})
-        prior_destructive_setting = await set_destructive_requires_name(es, False)
+
+        # bypassing action.destructive_requires_name cluster setting mangling for serverless clusters
+        prior_destructive_setting = None
+        cluster_settings_available = False
+        with contextlib.suppress(elasticsearch.exceptions.NotFoundError):
+            prior_destructive_setting = await set_destructive_requires_name(es, False)
+            cluster_settings_available = True
+
         try:
             for index_name in indices:
                 if not only_if_exists:
@@ -1425,7 +1436,8 @@ async def __call__(self, es, params):
                     await es.indices.delete(index=index_name, params=request_params)
                     ops += 1
         finally:
-            await set_destructive_requires_name(es, prior_destructive_setting)
+            if cluster_settings_available:
+                await set_destructive_requires_name(es, prior_destructive_setting)
         return {
             "weight": ops,
             "unit": "ops",

esrally/telemetry.py

@@ -103,6 +103,9 @@ def on_benchmark_start(self):
     def on_benchmark_stop(self):
         for device in self.devices:
             if self._enabled(device):
+                if self.serverless_mode and not self._available_on_serverless(device):
+                    # Not informing the user the second time, see on_benchmark_start()
+                    continue
                 device.on_benchmark_stop()
 
     def store_system_metrics(self, node, metrics_store):