Repository: elastic/rally

Issue:
Set retry-on-timeout=true for telemetry device connections
As noticed in https://discuss.elastic.co/t/rally-track-report-analysis-follow-up/127266/17 , there are occasions where firewalls drop connections when there is no traffic and for benchmarks that take a long time to finish, [GcTimesSummary](https://github.com/elastic/rally/blob/3bd432054f6ab592a4b83ce90fe009eeb28f9754/esrally/mechanic/telemetry.py#L807) can timeout since it creates a connection at the beginning of the benchmark and collects again at the end of the benchmark.

After an internal discussion, we agreed that it makes sense to enable `retry-on-timeout=true` for the client connections used by telemetry devices. This will avoid an unnecessary duplication of requests performed through other connections, such as the ones created in [driver.py](https://github.com/elastic/rally/blob/3bd432054f6ab592a4b83ce90fe009eeb28f9754/esrally/driver/driver.py#L588), for example, used for bulking.

Discussion:
dliappis: Thanks for the review! I addressed the comments can you take another pass please?

Patch:
diff --git a/docs/telemetry.rst b/docs/telemetry.rst
index b0cd9849..d24f0d22 100644
--- a/docs/telemetry.rst
+++ b/docs/telemetry.rst
@@ -93,5 +93,6 @@ Supported telemetry parameters:
 * ``node-stats-include-thread-pools`` (default: ``true``): A boolean indicating whether thread pool stats should be included.
 * ``node-stats-include-buffer-pools`` (default: ``true``): A boolean indicating whether buffer pool stats should be included.
 * ``node-stats-include-breakers`` (default: ``true``): A boolean indicating whether circuit breaker stats should be included.
+* ``node-stats-include-mem`` (default: ``true``): A boolean indicating whether JVM heap stats should be included.
 * ``node-stats-include-network`` (default: ``true``): A boolean indicating whether network-related stats should be included.
 * ``node-stats-include-process`` (default: ``true``): A boolean indicating whether process cpu stats should be included.
diff --git a/esrally/mechanic/launcher.py b/esrally/mechanic/launcher.py
index b49adc02..aa567951 100644
--- a/esrally/mechanic/launcher.py
+++ b/esrally/mechanic/launcher.py
@@ -68,7 +68,9 @@ class ClusterLauncher:
         es = {}
         for cluster_name, cluster_hosts in all_hosts.items():
             all_client_options = self.cfg.opts("client", "options").all_client_options
-            cluster_client_options = all_client_options[cluster_name]
+            cluster_client_options = dict(all_client_options[cluster_name])
+            # Use retries to avoid aborts on long living connections for telemetry devices
+            cluster_client_options["retry-on-timeout"] = True
             es[cluster_name] = self.client_factory(cluster_hosts, cluster_client_options).create()
 
         es_default = es["default"]
diff --git a/esrally/mechanic/telemetry.py b/esrally/mechanic/telemetry.py
index 67b9e9a1..8c078ffc 100644
--- a/esrally/mechanic/telemetry.py
+++ b/esrally/mechanic/telemetry.py
@@ -319,6 +319,7 @@ class NodeStatsRecorder:
         self.include_breakers = telemetry_params.get("node-stats-include-breakers", True)
         self.include_network = telemetry_params.get("node-stats-include-network", True)
         self.include_process = telemetry_params.get("node-stats-include-process", True)
+        self.include_mem_stats = telemetry_params.get("node-stats-include-mem", True)
         self.client = client
         self.metrics_store = metrics_store
 
@@ -339,6 +340,8 @@ class NodeStatsRecorder:
                 self.record_circuit_breaker_stats(node_name, node_stats)
             if self.include_buffer_pools:
                 self.record_jvm_buffer_pool_stats(node_name, node_stats)
+            if self.include_mem_stats:
+                self.record_jvm_mem_stats(node_name, node_stats)
             if self.include_network:
                 self.record_network_stats(node_name, node_stats)
             if self.include_process:
@@ -383,6 +386,15 @@ class NodeStatsRecorder:
                                node_stats_metric_name=metric_name,
                                metric_value=metric_value)
 
+    def record_jvm_mem_stats(self, node_name, node_stats):
+        mem_stats = node_stats["jvm"]["mem"]
+        if mem_stats:
+            for metric_name, metric_value in mem_stats.items():
+                self.put_value(node_name,
+                               metric_name="jvm_mem_{}".format(metric_name),
+                               node_stats_metric_name=metric_name,
+                               metric_value=metric_value)
+
     def record_network_stats(self, node_name, node_stats):
         transport_stats = node_stats.get("transport")
         if transport_stats:


Test patch:
diff --git a/tests/mechanic/launcher_test.py b/tests/mechanic/launcher_test.py
index 0d7687ea..9b60004f 100644
--- a/tests/mechanic/launcher_test.py
+++ b/tests/mechanic/launcher_test.py
@@ -141,6 +141,19 @@ class ClusterLauncherTests(TestCase):
         self.assertEqual([{"host": "10.0.0.10", "port":9200}, {"host": "10.0.0.11", "port":9200}], cluster.hosts)
         self.assertIsNotNone(cluster.telemetry)
 
+    def test_launches_cluster_with_telemetry_client_timeout_enabled(self):
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", self.test_host)
+        cfg.add(config.Scope.application, "client", "options", self.client_options)
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
+        cluster = cluster_launcher.start()
+
+        for telemetry_device in cluster.telemetry.devices:
+            self.assertDictEqual({"retry-on-timeout": True, "timeout": 60}, telemetry_device.client.client_options)
+
     @mock.patch("time.sleep")
     def test_error_on_cluster_launch(self, sleep):
         on_post_launch = mock.Mock()
diff --git a/tests/mechanic/telemetry_test.py b/tests/mechanic/telemetry_test.py
index caa22c24..af714f70 100644
--- a/tests/mechanic/telemetry_test.py
+++ b/tests/mechanic/telemetry_test.py
@@ -304,6 +304,34 @@ class NodeStatsRecorderTests(TestCase):
                             "current_loaded_count" : 9992,
                             "total_loaded_count" : 9992,
                             "total_unloaded_count" : 0
+                        },
+                        "mem": {
+                            "heap_used_in_bytes": 119073552,
+                            "heap_used_percent": 19,
+                            "heap_committed_in_bytes": 626393088,
+                            "heap_max_in_bytes": 626393088,
+                            "non_heap_used_in_bytes": 110250424,
+                            "non_heap_committed_in_bytes": 118108160,
+                            "pools": {
+                                "young": {
+                                    "used_in_bytes": 66378576,
+                                    "max_in_bytes": 139591680,
+                                    "peak_used_in_bytes": 139591680,
+                                    "peak_max_in_bytes": 139591680
+                                },
+                                "survivor": {
+                                    "used_in_bytes": 358496,
+                                    "max_in_bytes": 17432576,
+                                    "peak_used_in_bytes": 17432576,
+                                    "peak_max_in_bytes": 17432576
+                                },
+                                "old": {
+                                    "used_in_bytes": 52336480,
+                                    "max_in_bytes": 469368832,
+                                    "peak_used_in_bytes": 52336480,
+                                    "peak_max_in_bytes": 469368832
+                                }
+                            }
                         }
                     },
                     "process": {
@@ -467,6 +495,34 @@ class NodeStatsRecorderTests(TestCase):
                             "current_loaded_count" : 9992,
                             "total_loaded_count" : 9992,
                             "total_unloaded_count" : 0
+                        },
+                        "mem": {
+                            "heap_used_in_bytes": 119073552,
+                            "heap_used_percent": 19,
+                            "heap_committed_in_bytes": 626393088,
+                            "heap_max_in_bytes": 626393088,
+                            "non_heap_used_in_bytes": 110250424,
+                            "non_heap_committed_in_bytes": 118108160,
+                            "pools": {
+                                "young": {
+                                    "used_in_bytes": 66378576,
+                                    "max_in_bytes": 139591680,
+                                    "peak_used_in_bytes": 139591680,
+                                    "peak_max_in_bytes": 139591680
+                                },
+                                "survivor": {
+                                    "used_in_bytes": 358496,
+                                    "max_in_bytes": 17432576,
+                                    "peak_used_in_bytes": 17432576,
+                                    "peak_max_in_bytes": 17432576
+                                },
+                                "old": {
+                                    "used_in_bytes": 52336480,
+                                    "max_in_bytes": 469368832,
+                                    "peak_used_in_bytes": 52336480,
+                                    "peak_max_in_bytes": 469368832
+                                }
+                            }
                         }
                     },
                     "process": {
@@ -556,6 +612,7 @@ class NodeStatsRecorderTests(TestCase):
             mock.call(node_name="rally0", name="breaker_parent_tripped", count=0),
             mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_count", count=7),
             mock.call(node_name="rally0", name="jvm_buffer_pool_direct_count", count=6),
+            mock.call(node_name="rally0", name="jvm_mem_heap_used_percent", count=19),
         ], any_order=True)
 
         metrics_store_put_value.assert_has_calls([
@@ -578,6 +635,11 @@ class NodeStatsRecorderTests(TestCase):
             mock.call(node_name="rally0", name="jvm_buffer_pool_mapped_total_capacity_in_bytes", value=9999, unit="byte"),
             mock.call(node_name="rally0", name="jvm_buffer_pool_direct_used_in_bytes", value=73868, unit="byte"),
             mock.call(node_name="rally0", name="jvm_buffer_pool_direct_total_capacity_in_bytes", value=73867, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_mem_heap_used_in_bytes", value=119073552, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_mem_heap_committed_in_bytes", value=626393088, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_mem_heap_max_in_bytes", value=626393088, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_mem_non_heap_used_in_bytes", value=110250424, unit="byte"),
+            mock.call(node_name="rally0", name="jvm_mem_non_heap_committed_in_bytes", value=118108160, unit="byte"),
         ], any_order=True)