Add master node telemetry device

docs/telemetry.rst

@@ -226,4 +226,20 @@ Example of recorded documents given two data streams in the cluster::
 
 Supported telemetry parameters:
 
-* ``data-stream-stats-sample-interval`` (default 10): A positive number greater than zero denoting the sampling interval in seconds.
+* ``data-stream-stats-sample-interval`` (default 10): A positive number greater than zero denoting the sampling interval in seconds.
+
+
+master-node-stats
+-----------------
+
+The master-node-stats telemetry device regularly reports the name of the master node by using:
+
+ 1. the `Cluster State API <https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-state.html>`_, specifically `/_cluster/state/master_node` to determine the id of the master node
+ 2. the `Nodes info API <https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-nodes-info.html>`_ to determine its name
+
+Example of recorded document::
+
+   {
+     "name": "master-node-stats",
+     "node": "rally-0"
+   }

esrally/driver/driver.py

@@ -603,6 +603,7 @@ def prepare_telemetry(self, es, enable):
                 telemetry.JvmStatsSummary(es_default, self.metrics_store),
                 telemetry.IndexStats(es_default, self.metrics_store),
                 telemetry.MlBucketProcessingTime(es_default, self.metrics_store),
+                telemetry.MasterNodeStats(telemetry_params, es_default, self.metrics_store),
                 telemetry.SegmentStats(log_root, es_default),
                 telemetry.CcrStats(telemetry_params, es, self.metrics_store),
                 telemetry.RecoveryStats(telemetry_params, es, self.metrics_store),

esrally/telemetry.py

@@ -1926,3 +1926,91 @@ def detach_from_node(self, node, running):
     def store_system_metrics(self, node, metrics_store):
         if self.index_size_bytes:
             metrics_store.put_value_node_level(node.node_name, "final_index_size_bytes", self.index_size_bytes, "byte")
+
+
+class MasterNodeStats(InternalTelemetryDevice):
+    """
+    Collects and pushes the current master node name to the metric store.
+    """
+
+    command = "master-node-stats"
+    human_name = "Master Node Stats"
+    help = "Regularly samples master node name"
+
+    def __init__(self, telemetry_params, client, metrics_store):
+        """
+        :param telemetry_params: The configuration object for telemetry_params.
+            May optionally specify:
+            ``master-node-stats-sample-interval``: An integer controlling the interval, in seconds,
+            between collecting samples. Default: 30s.
+        :param client: The default Elasticsearch client
+        :param metrics_store: The configured metrics store we write to.
+        """
+        super().__init__()
+
+        self.telemetry_params = telemetry_params
+        self.client = client
+        self.sample_interval = telemetry_params.get("master-node-stats-sample-interval", 30)
+        if self.sample_interval <= 0:
+            raise exceptions.SystemSetupError(
+                f"The telemetry parameter 'master-node-stats-sample-interval' must be greater than zero " f"but was {self.sample_interval}."
+            )
+        self.metrics_store = metrics_store
+        self.sampler = None
+
+    def on_benchmark_start(self):
+        recorder = MasterNodeStatsRecorder(
+            self.client,
+            self.metrics_store,
+            self.sample_interval,
+        )
+        self.sampler = SamplerThread(recorder)
+        self.sampler.daemon = True
+        # we don't require starting recorders precisely at the same time
+        self.sampler.start()
+
+    def on_benchmark_stop(self):
+        if self.sampler:
+            self.sampler.finish()
+
+
+class MasterNodeStatsRecorder:
+    """
+    Collects and pushes the current master node name for the specified cluster to the metric store.
+    """
+
+    def __init__(self, client, metrics_store, sample_interval):
+        """
+        :param client: The Elasticsearch client for this cluster.
+        :param metrics_store: The configured metrics store we write to.
+        :param sample_interval: An integer controlling the interval, in seconds, between collecting samples.
+        """
+
+        self.client = client
+        self.metrics_store = metrics_store
+        self.sample_interval = sample_interval
+        self.logger = logging.getLogger(__name__)
+
+    def __str__(self):
+        return "master node stats"
+
+    def record(self):
+        """
+        Collect master node name and push to metrics store.
+        """
+        # pylint: disable=import-outside-toplevel
+        import elasticsearch
+
+        try:
+            info = self.client.cat.master(format="json")
+        except elasticsearch.TransportError:
+            msg = f"A transport error occurred while collecting master node stats on cluster [{self.cluster_name}]"
+            self.logger.exception(msg)
+            raise exceptions.RallyError(msg)
+
+        doc = {
+            "name": "master-node-stats",
+            "node": info[0]["node"],
+        }
+
+        self.metrics_store.put_doc(doc, level=MetaInfoScope.cluster)

tests/driver/driver_test.py

@@ -77,6 +77,7 @@ def __init__(self, *args, **kwargs):
             DriverTests.StaticClientFactory.PATCHER = mock.patch("elasticsearch.Elasticsearch")
             self.es = DriverTests.StaticClientFactory.PATCHER.start()
             self.es.indices.stats.return_value = {"mocked": True}
+            self.es.cat.master.return_value = {"mocked": True}
 
         def create(self):
             return self.es

tests/telemetry_test.py

@@ -103,11 +103,12 @@ def test_store_calculated_metrics(self, metrics_store_put_value, stop_watch):
 
 
 class Client:
-    def __init__(self, nodes=None, info=None, indices=None, transform=None, transport_client=None):
+    def __init__(self, nodes=None, info=None, indices=None, transform=None, cat=None, transport_client=None):
         self.nodes = nodes
         self._info = wrap(info)
         self.indices = indices
         self.transform = transform
+        self.cat = cat
         if transport_client:
             self.transport = transport_client
 
@@ -116,12 +117,13 @@ def info(self):
 
 
 class SubClient:
-    def __init__(self, stats=None, info=None, recovery=None, transform_stats=None, data_streams_stats=None):
+    def __init__(self, stats=None, info=None, recovery=None, transform_stats=None, data_streams_stats=None, master=None):
         self._stats = wrap(stats)
         self._info = wrap(info)
         self._recovery = wrap(recovery)
         self._transform_stats = wrap(transform_stats)
         self._data_streams_stats = wrap(data_streams_stats)
+        self._master = wrap(master)
 
     def stats(self, *args, **kwargs):
         return self._stats()
@@ -138,6 +140,9 @@ def get_transform_stats(self, *args, **kwargs):
     def data_streams_stats(self, *args, **kwargs):
         return self._data_streams_stats()
 
+    def master(self, *args, **kwargs):
+        return self._master()
+
 
 def wrap(it):
     return it if callable(it) else ResponseSupplier(it)
@@ -3664,3 +3669,43 @@ def test_stores_nothing_if_no_data_path(self, run_subprocess, metrics_store_clus
         self.assertEqual(0, run_subprocess.call_count)
         self.assertEqual(0, metrics_store_cluster_value.call_count)
         self.assertEqual(0, get_size.call_count)
+
+
+class MasterNodeStatsTests(TestCase):
+    def test_negative_sample_interval_forbidden(self):
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        telemetry_params = {"master-node-stats-sample-interval": -1}
+        with self.assertRaisesRegex(
+            exceptions.SystemSetupError,
+            r"The telemetry parameter 'master-node-stats-sample-interval' must be greater than zero but was .*\.",
+        ):
+            telemetry.MasterNodeStats(telemetry_params, Client(), metrics_store)
+
+
+class MasterNodeStatsRecorderTests(TestCase):
+    master_node_cat_response = [
+        {
+            "node": "rally-0",
+        }
+    ]
+
+    @mock.patch("esrally.metrics.EsMetricsStore.put_doc")
+    def test_store_master_node_stats(self, metrics_store_put_doc):
+        client = Client(cat=SubClient(master=self.master_node_cat_response))
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        recorder = telemetry.MasterNodeStatsRecorder(client=client, metrics_store=metrics_store, sample_interval=1)
+        recorder.record()
+
+        metrics_store_put_doc.assert_has_calls(
+            [
+                mock.call(
+                    {
+                        "name": "master-node-stats",
+                        "node": "rally-0",
+                    },
+                    level=MetaInfoScope.cluster,
+                ),
+            ],
+        )