Detect serverless operator status automatically (#1768)

Author: pquentin

esrally/client/factory.py

@@ -332,12 +332,13 @@ def cluster_distribution_version(hosts, client_options, client_factory=EsClientF
     :param client_options: The client options to customize the Elasticsearch client.
     :param client_factory: Factory class that creates the Elasticsearch client.
     :return: The cluster's build flavor, version number, and build hash. For Serverless Elasticsearch these may all be
-      the build flavor value.
+      the build flavor value. Also returns the operator status (always False for stateful).
     """
     # no way for us to know whether we're talking to a serverless elasticsearch or not, so we default to the sync client
     es = client_factory(hosts, client_options).create()
-    # unconditionally wait for the REST layer - if it's not up by then, we'll intentionally raise the original error
-    wait_for_rest_layer(es)
+
+    # wait_for_rest_layer  calls the Cluster Health API, which is not available for unprivileged users on Serverless
+    # As a result, we need to call the info API first to know if we can call wait_for_rest_layer().
     version = es.info()["version"]
 
     version_build_flavor = version.get("build_flavor", "oss")
@@ -346,7 +347,16 @@ def cluster_distribution_version(hosts, client_options, client_factory=EsClientF
     # version number does not exist for serverless
     version_number = version.get("number", version_build_flavor)
 
-    return version_build_flavor, version_number, version_build_hash
+    serverless_operator = False
+    if versions.is_serverless(version_build_flavor):
+        authentication_info = es.perform_request(method="GET", path="/_security/_authenticate")
+        serverless_operator = authentication_info.body.get("operator", False)
+
+    if not versions.is_serverless(version_build_flavor) or serverless_operator is True:
+        # if available, unconditionally wait for the REST layer - if it's not up, we'll intentionally raise the original error
+        wait_for_rest_layer(es)
+
+    return version_build_flavor, version_number, version_build_hash, serverless_operator
 
 
 def create_api_key(es, client_id, max_attempts=5):

esrally/metrics.py

@@ -264,7 +264,9 @@ def __init__(self, cfg):
 
         # TODO #1335: Use version-specific support for metrics stores after 7.8.0.
         if self.probe_version:
-            distribution_flavor, distribution_version, _ = client.cluster_distribution_version(hosts=hosts, client_options=client_options)
+            distribution_flavor, distribution_version, _, _ = client.cluster_distribution_version(
+                hosts=hosts, client_options=client_options
+            )
             self._cluster_version = distribution_version
 
         factory = client.EsClientFactory(

esrally/racecontrol.py

@@ -186,9 +186,12 @@ def setup(self, sources=False):
         if not sources and not self.cfg.exists("mechanic", "distribution.version"):
             hosts = self.cfg.opts("client", "hosts").default
             client_options = self.cfg.opts("client", "options").default
-            distribution_flavor, distribution_version, distribution_build_hash = client.factory.cluster_distribution_version(
-                hosts, client_options
-            )
+            (
+                distribution_flavor,
+                distribution_version,
+                distribution_build_hash,
+                serverless_operator,
+            ) = client.factory.cluster_distribution_version(hosts, client_options)
 
             self.logger.info(
                 "Automatically derived distribution flavor [%s], version [%s], and build hash [%s]",
@@ -203,8 +206,8 @@ def setup(self, sources=False):
                     self.cfg.add(config.Scope.benchmark, "driver", "serverless.mode", True)
 
                 if not self.cfg.exists("driver", "serverless.operator"):
-                    # operator privileges assumed for now
-                    self.cfg.add(config.Scope.benchmark, "driver", "serverless.operator", True)
+                    self.cfg.add(config.Scope.benchmark, "driver", "serverless.operator", serverless_operator)
+                console.info(f"Detected Elasticsearch Serverless mode with operator=[{serverless_operator}].")
             else:
                 min_es_version = versions.Version.from_string(version.minimum_es_version())
                 specified_version = versions.Version.from_string(distribution_version)

esrally/tracker/tracker.py

@@ -78,7 +78,7 @@ def create_track(cfg):
     client_options = cfg.opts("client", "options").default
     data_streams = cfg.opts("generator", "data_streams")
 
-    distribution_flavor, distribution_version, _ = factory.cluster_distribution_version(target_hosts, client_options)
+    distribution_flavor, distribution_version, _, _ = factory.cluster_distribution_version(target_hosts, client_options)
     client = factory.EsClientFactory(
         hosts=target_hosts,
         client_options=client_options,