Fetch and store target cluster meta info at benchmark start

esrally/driver/driver.py

@@ -134,10 +134,13 @@ class WorkerIdle:
 
 
 class PreparationComplete:
-    def __init__(self, distribution_flavor, distribution_version, revision):
+    def __init__(self, distribution_flavor, distribution_version, revision, target_id=None, target_platform=None, target_auth_type=None):
         self.distribution_flavor = distribution_flavor
         self.distribution_version = distribution_version
         self.revision = revision
+        self.target_id = target_id
+        self.target_platform = target_platform
+        self.target_auth_type = target_auth_type
 
 
 class StartWorker:
@@ -235,6 +238,8 @@ def __init__(self):
         self.status = "init"
         self.post_process_timer = 0
         self.cluster_details = {}
+        self.target_platform = None
+        self.target_auth_type = None
 
     def receiveMsg_PoisonMessage(self, poisonmsg, sender):
         self.logger.error("Main driver received a fatal indication from a load generator (%s). Shutting down.", poisonmsg.details)
@@ -359,6 +364,9 @@ def _after_track_prepared(self):
         build_version = cluster_version.get("number", build_flavor)
         build_hash = cluster_version.get("build_hash", build_flavor)
 
+        # Determine target_id (cluster_name from GET /)
+        target_id = self.cluster_details.get("cluster_name") if self.cluster_details else None
+
         for child in self.children:
             self.send(child, thespian.actors.ActorExitRequest())
         self.children = []
@@ -368,6 +376,9 @@ def _after_track_prepared(self):
                 build_flavor,
                 build_version,
                 build_hash,
+                target_id=target_id,
+                target_platform=self.target_platform,
+                target_auth_type=self.target_auth_type,
             ),
         )
 
@@ -631,12 +642,14 @@ def prepare_telemetry(self, es, enable, index_names, data_stream_names, build_ha
         log_root = paths.race_root(self.config)
 
         es_default = es["default"]
+        all_client_options = self.config.opts("client", "options").all_client_options
+        default_client_options = all_client_options.get("default", {})
 
         if enable:
             devices = [
                 telemetry.NodeStats(telemetry_params, es, self.metrics_store),
                 telemetry.ExternalEnvironmentInfo(es_default, self.metrics_store),
-                telemetry.ClusterEnvironmentInfo(es_default, self.metrics_store, build_hash),
+                telemetry.ClusterEnvironmentInfo(es_default, self.metrics_store, build_hash, default_client_options),
                 telemetry.JvmStatsSummary(es_default, self.metrics_store),
                 telemetry.IndexStats(es_default, self.metrics_store),
                 telemetry.MlBucketProcessingTime(es_default, self.metrics_store),
@@ -739,6 +752,27 @@ def prepare_benchmark(self, t):
                     self.logger.info("Retrieved actual build hash [%s] from serverless cluster.", build_hash)
                     self.driver_actor.cluster_details["version"]["build_hash"] = build_hash
 
+            # Determine target_platform while we still have the API response object (with response headers)
+            if serverless_mode:
+                self.driver_actor.target_platform = "serverless"
+            else:
+                target_platform = "on-prem"
+                try:
+                    cluster_details = self.driver_actor.cluster_details
+                    meta = getattr(cluster_details, "meta", None)
+                    if meta is not None and "x-found-handling-cluster" in getattr(meta, "headers", {}):
+                        target_platform = "hosted"
+                except Exception:
+                    pass
+                self.driver_actor.target_platform = target_platform
+
+            # Determine target_auth_type from default client options
+            default_client_options = self.config.opts("client", "options").all_client_options.get("default", {})
+            if default_client_options.get("api_key"):
+                self.driver_actor.target_auth_type = "api_key"
+            elif default_client_options.get("basic_auth_user") or default_client_options.get("basic_auth"):
+                self.driver_actor.target_auth_type = "basic"
+
         # Avoid issuing any requests to the target cluster when static responses are enabled. The results
         # are not useful and attempts to connect to a non-existing cluster just lead to exception traces in logs.
         self.prepare_telemetry(

esrally/metrics.py

@@ -1462,6 +1462,9 @@ def __init__(
         revision=None,
         results=None,
         meta_data=None,
+        target_id=None,
+        target_platform=None,
+        target_auth_type=None,
     ):
         if results is None:
             results = {}
@@ -1492,6 +1495,9 @@ def __init__(
         self.revision = revision
         self.results = results
         self.meta_data = meta_data
+        self.target_id = target_id
+        self.target_platform = target_platform
+        self.target_auth_type = target_auth_type
 
     @property
     def track_name(self):
@@ -1578,6 +1584,12 @@ def to_result_dicts(self):
             result_template["car-params"] = self.car_params
         if self.plugin_params:
             result_template["plugin-params"] = self.plugin_params
+        if self.target_id:
+            result_template["target-id"] = self.target_id
+        if self.target_platform:
+            result_template["target-platform"] = self.target_platform
+        if self.target_auth_type:
+            result_template["target-auth-type"] = self.target_auth_type
         if self.meta_data:
             result_template["meta"] = self.meta_data
 

esrally/racecontrol.py

@@ -137,7 +137,14 @@ def receiveMsg_EngineStarted(self, msg, sender):
 
     @actor.no_retry("race control")  # pylint: disable=no-value-for-parameter
     def receiveMsg_PreparationComplete(self, msg, sender):
-        self.coordinator.on_preparation_complete(msg.distribution_flavor, msg.distribution_version, msg.revision)
+        self.coordinator.on_preparation_complete(
+            msg.distribution_flavor,
+            msg.distribution_version,
+            msg.revision,
+            target_id=msg.target_id,
+            target_platform=msg.target_platform,
+            target_auth_type=msg.target_auth_type,
+        )
         self.logger.info("Telling driver to start benchmark.")
         self.send(self.main_driver, driver.StartBenchmark())
 
@@ -246,10 +253,18 @@ def setup(self, sources=False):
         )
         self.race_store = metrics.race_store(self.cfg)
 
-    def on_preparation_complete(self, distribution_flavor, distribution_version, revision):
+    def on_preparation_complete(
+        self, distribution_flavor, distribution_version, revision, target_id=None, target_platform=None, target_auth_type=None
+    ):
         self.race.distribution_flavor = distribution_flavor
         self.race.distribution_version = distribution_version
         self.race.revision = revision
+        if target_id is not None:
+            self.race.target_id = target_id
+        if target_platform is not None:
+            self.race.target_platform = target_platform
+        if target_auth_type is not None:
+            self.race.target_auth_type = target_auth_type
         # store race initially (without any results) so other components can retrieve full metadata
         self.race_store.store_race(self.race)
         if self.race.challenge.auto_generated:

esrally/telemetry.py

@@ -1809,11 +1809,12 @@ class ClusterEnvironmentInfo(InternalTelemetryDevice):
 
     serverless_status = serverless.Status.Public
 
-    def __init__(self, client, metrics_store, revision_override):
+    def __init__(self, client, metrics_store, revision_override, client_options=None):
         super().__init__()
         self.metrics_store = metrics_store
         self.client = client
         self.revision_override = revision_override
+        self.client_options = client_options if client_options is not None else {}
 
     def on_benchmark_start(self):
         # noinspection PyBroadException
@@ -1836,6 +1837,33 @@ def on_benchmark_start(self):
         self.metrics_store.add_meta_info(metrics.MetaInfoScope.cluster, None, "distribution_version", distribution_version)
         self.metrics_store.add_meta_info(metrics.MetaInfoScope.cluster, None, "distribution_flavor", distribution_flavor)
 
+        # Store target ID (project ID for serverless, cluster ID for ECH, cluster name for on-prem)
+        target_id = client_info.get("cluster_name")
+        if target_id:
+            self.metrics_store.add_meta_info(metrics.MetaInfoScope.cluster, None, "target_id", target_id)
+
+        # Determine and store target platform
+        if versions.is_serverless(distribution_flavor):
+            target_platform = "serverless"
+        else:
+            # Check for ECH (Elastic Cloud Hosted) by looking for cloud-specific response headers
+            target_platform = "on-prem"
+            try:
+                meta = getattr(client_info, "meta", None)
+                if meta is not None:
+                    headers = getattr(meta, "headers", {})
+                    if "x-found-handling-cluster" in headers:
+                        target_platform = "hosted"
+            except Exception:
+                pass
+        self.metrics_store.add_meta_info(metrics.MetaInfoScope.cluster, None, "target_platform", target_platform)
+
+        # Determine and store auth type from client options
+        if self.client_options.get("api_key"):
+            self.metrics_store.add_meta_info(metrics.MetaInfoScope.cluster, None, "target_auth_type", "api_key")
+        elif self.client_options.get("basic_auth_user") or self.client_options.get("basic_auth"):
+            self.metrics_store.add_meta_info(metrics.MetaInfoScope.cluster, None, "target_auth_type", "basic")
+
 
 def add_metadata_for_node(metrics_store, node_name, host_name):
     """

tests/telemetry_test.py

@@ -3463,6 +3463,7 @@ def test_stores_cluster_level_metrics_on_attach(self, metrics_store_add_meta_inf
             mock.call(metrics.MetaInfoScope.cluster, None, "source_revision", "abc123"),
             mock.call(metrics.MetaInfoScope.cluster, None, "distribution_version", "6.0.0-alpha1"),
             mock.call(metrics.MetaInfoScope.cluster, None, "distribution_flavor", "oss"),
+            mock.call(metrics.MetaInfoScope.cluster, None, "target_platform", "on-prem"),
         ]
 
         metrics_store_add_meta_info.assert_has_calls(calls)
@@ -3498,6 +3499,7 @@ def test_static_serverless_version_overwrite(self, metrics_store_add_meta_info):
             mock.call(metrics.MetaInfoScope.cluster, None, "source_revision", "00000000"),
             mock.call(metrics.MetaInfoScope.cluster, None, "distribution_version", "serverless"),
             mock.call(metrics.MetaInfoScope.cluster, None, "distribution_flavor", "serverless"),
+            mock.call(metrics.MetaInfoScope.cluster, None, "target_platform", "serverless"),
         ]
 
         metrics_store_add_meta_info.assert_has_calls(calls)
@@ -3522,10 +3524,153 @@ def test_revision_override(self, metrics_store_add_meta_info):
             mock.call(metrics.MetaInfoScope.cluster, None, "source_revision", "abc123"),
             mock.call(metrics.MetaInfoScope.cluster, None, "distribution_version", "serverless"),
             mock.call(metrics.MetaInfoScope.cluster, None, "distribution_flavor", "serverless"),
+            mock.call(metrics.MetaInfoScope.cluster, None, "target_platform", "serverless"),
         ]
 
         metrics_store_add_meta_info.assert_has_calls(calls)
 
+    @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
+    def test_stores_cluster_name(self, metrics_store_add_meta_info):
+        cluster_info = {
+            "cluster_name": "my-benchmark-cluster",
+            "version": {
+                "build_hash": "abc123",
+                "number": "8.0.0",
+                "build_flavor": "default",
+            },
+        }
+
+        cfg = create_config()
+        client = Client(info=cluster_info)
+        metrics_store = metrics.EsMetricsStore(cfg)
+        env_device = telemetry.ClusterEnvironmentInfo(client, metrics_store, None)
+        t = telemetry.Telemetry(cfg, devices=[env_device])
+        t.on_benchmark_start()
+        calls = [
+            mock.call(metrics.MetaInfoScope.cluster, None, "target_id", "my-benchmark-cluster"),
+            mock.call(metrics.MetaInfoScope.cluster, None, "target_platform", "on-prem"),
+        ]
+
+        metrics_store_add_meta_info.assert_has_calls(calls)
+
+    @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
+    def test_detects_hosted_platform_from_response_headers(self, metrics_store_add_meta_info):
+        class ApiResponseMeta:
+            def __init__(self, headers):
+                self.headers = headers
+
+        class ApiResponse:
+            def __init__(self, data, headers):
+                self._data = data
+                self.meta = ApiResponseMeta(headers)
+
+            def __getitem__(self, key):
+                return self._data[key]
+
+            def get(self, key, default=None):
+                return self._data.get(key, default)
+
+        cluster_info = ApiResponse(
+            {
+                "cluster_name": "my-ech-cluster",
+                "version": {
+                    "build_hash": "abc123",
+                    "number": "8.0.0",
+                    "build_flavor": "default",
+                },
+            },
+            headers={"x-found-handling-cluster": "abc123def456"},
+        )
+
+        cfg = create_config()
+        client = Client(info=cluster_info)
+        metrics_store = metrics.EsMetricsStore(cfg)
+        env_device = telemetry.ClusterEnvironmentInfo(client, metrics_store, None)
+        t = telemetry.Telemetry(cfg, devices=[env_device])
+        t.on_benchmark_start()
+
+        metrics_store_add_meta_info.assert_any_call(metrics.MetaInfoScope.cluster, None, "target_platform", "hosted")
+
+    @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
+    def test_stores_api_key_auth_type(self, metrics_store_add_meta_info):
+        cluster_info = {
+            "version": {
+                "build_hash": "abc123",
+                "number": "8.0.0",
+                "build_flavor": "default",
+            },
+        }
+
+        cfg = create_config()
+        client = Client(info=cluster_info)
+        metrics_store = metrics.EsMetricsStore(cfg)
+        client_options = {"api_key": "my-api-key", "timeout": 60}
+        env_device = telemetry.ClusterEnvironmentInfo(client, metrics_store, None, client_options)
+        t = telemetry.Telemetry(cfg, devices=[env_device])
+        t.on_benchmark_start()
+
+        metrics_store_add_meta_info.assert_any_call(metrics.MetaInfoScope.cluster, None, "target_auth_type", "api_key")
+
+    @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
+    def test_stores_basic_auth_type(self, metrics_store_add_meta_info):
+        cluster_info = {
+            "version": {
+                "build_hash": "abc123",
+                "number": "8.0.0",
+                "build_flavor": "default",
+            },
+        }
+
+        cfg = create_config()
+        client = Client(info=cluster_info)
+        metrics_store = metrics.EsMetricsStore(cfg)
+        client_options = {"basic_auth_user": "elastic", "basic_auth_password": "changeme", "timeout": 60}
+        env_device = telemetry.ClusterEnvironmentInfo(client, metrics_store, None, client_options)
+        t = telemetry.Telemetry(cfg, devices=[env_device])
+        t.on_benchmark_start()
+
+        metrics_store_add_meta_info.assert_any_call(metrics.MetaInfoScope.cluster, None, "target_auth_type", "basic")
+
+    @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
+    def test_stores_basic_auth_type_from_combined_tuple(self, metrics_store_add_meta_info):
+        cluster_info = {
+            "version": {
+                "build_hash": "abc123",
+                "number": "8.0.0",
+                "build_flavor": "default",
+            },
+        }
+
+        cfg = create_config()
+        client = Client(info=cluster_info)
+        metrics_store = metrics.EsMetricsStore(cfg)
+        client_options = {"basic_auth": ("elastic", "changeme"), "timeout": 60}
+        env_device = telemetry.ClusterEnvironmentInfo(client, metrics_store, None, client_options)
+        t = telemetry.Telemetry(cfg, devices=[env_device])
+        t.on_benchmark_start()
+
+        metrics_store_add_meta_info.assert_any_call(metrics.MetaInfoScope.cluster, None, "target_auth_type", "basic")
+
+    @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")
+    def test_no_auth_type_stored_when_no_auth_options(self, metrics_store_add_meta_info):
+        cluster_info = {
+            "version": {
+                "build_hash": "abc123",
+                "number": "8.0.0",
+                "build_flavor": "default",
+            },
+        }
+
+        cfg = create_config()
+        client = Client(info=cluster_info)
+        metrics_store = metrics.EsMetricsStore(cfg)
+        env_device = telemetry.ClusterEnvironmentInfo(client, metrics_store, None)
+        t = telemetry.Telemetry(cfg, devices=[env_device])
+        t.on_benchmark_start()
+
+        auth_type_calls = [c for c in metrics_store_add_meta_info.call_args_list if len(c[0]) >= 3 and c[0][2] == "target_auth_type"]
+        assert len(auth_type_calls) == 0
+
 
 class TestNodeEnvironmentInfo:
     @mock.patch("esrally.metrics.EsMetricsStore.add_meta_info")