Allow to forcefully close SSL connections (#1016)

With this commit we expose a new client option `enable_cleanup_closed`
which allows to forcefully close an SSL connection that might not have
been properly closed on the server-side. This avoids potentially leaking
connections.

docs/command_line_reference.rst

@@ -644,6 +644,7 @@ Default value: ``timeout:60``
 Rally recognizes the following client options in addition:
 
 * ``max_connections``: By default, Rally will choose the maximum allowed number of connections automatically (equal to the number of simulated clients but at least 256 connections). With this property it is possible to override that logic but a minimum of 256 is enforced internally.
+* ``enable_cleanup_closed`` (default: ``false``): In some cases, SSL connections might not be properly closed and the number of open connections increases as a result. When this client option is set to ``true``, the Elasticsearch client will check and forcefully close these connections.
 
 **Examples**
 

esrally/async_connection.py

@@ -83,6 +83,7 @@ def __init__(self, host='localhost', port=9200, http_auth=None,
 
         trace_configs = [trace_config] if trace_config else None
         max_connections = max(256, kwargs.get("max_connections", 0))
+        enable_cleanup_closed = kwargs.get("enable_cleanup_closed", False)
         self.session = aiohttp.ClientSession(
             auth=http_auth,
             timeout=self.timeout,
@@ -91,7 +92,8 @@ def __init__(self, host='localhost', port=9200, http_auth=None,
                 verify_ssl=verify_certs,
                 use_dns_cache=use_dns_cache,
                 ssl_context=ssl_context,
-                limit=max_connections
+                limit=max_connections,
+                enable_cleanup_closed=enable_cleanup_closed
             ),
             headers=headers,
             trace_configs=trace_configs,

esrally/client.py

@@ -23,7 +23,7 @@
 import urllib3
 
 from esrally import exceptions, doc_link
-from esrally.utils import console
+from esrally.utils import console, convert
 
 
 class EsClientFactory:
@@ -118,6 +118,9 @@ def __init__(self, hosts, client_options):
         else:
             self.logger.info("HTTP compression: off")
 
+        if self._is_set(self.client_options, "enable_cleanup_closed"):
+            self.client_options["enable_cleanup_closed"] = convert.to_bool(self.client_options.pop("enable_cleanup_closed"))
+
     def _is_set(self, client_opts, k):
         try:
             return client_opts[k]