Repository: elastic/rally

Issue:
Add new telemetry device: recovery-stats
We need a new telemetry device that can collect the data available in `_recovery`, see [docs](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-recovery.html).

The `_recovery` stats somewhat resemble the structure of [ccr stats](https://www.elastic.co/guide/en/elasticsearch/reference/current/ccr-get-stats.html) i.e. using the form

```
{
  "index1" : {
    "shards" : [ {
      "id" : 0,
      "type" : "SNAPSHOT",
      "stage" : "INDEX",
```

(note however that ccr stats looks like:
```
"follow_stats" : {
    "indices" : [
      {
        "index" : "follower_index",
        "shards" : [
          {
            "remote_cluster" : "remote_cluster",
            ...
```

so the ccr stats telemetry device code can't be used as is)

It would be great if we can also support a telemetry device option for `detailed: true` as described in the [_recovery docs](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-recovery.html).

Discussion:
danielmitterdorfer: I've tested the changes with the attached [test-track](https://github.com/elastic/rally/files/2812511/test-track.zip). That track restores an index `taxis` from a snapshot.

Test scenarios (all with an ES metrics store):

* `esrally --pipeline=from-sources-skip-build --telemetry=recovery-stats --track-path=/path/to/test-track`
* `esrally --pipeline=from-sources-skip-build --telemetry=recovery-stats --track-path=/path/to/test-track --telemetry-params="recovery-stats-indices:'ta*'"`
* `esrally --pipeline=from-sources-skip-build --telemetry=recovery-stats --track-path=/path/to/test-track --telemetry-params="recovery-stats-indices:'bla*'"`
* `esrally --pipeline=from-sources-skip-build --telemetry=recovery-stats --track-path=/path/to/test-track --telemetry-params="recovery-stats-indices:'ta*',recovery-stats-sample-interval:2"`
* `esrally --pipeline=from-sources-skip-build --telemetry=recovery-stats --track-path=/path/to/test-track --telemetry-params="telemetry-params.json"`

`telemetry-params.json`:

```json
{
  "recovery-stats-indices": {
    "default": "taxi*"
  },
  "recovery-stats-sample-interval": 2
}
```


danielmitterdorfer: I've tested this also with the in-memory metrics store and although this telemetry device is not useful there (it only writes raw metrics entries which get lost if the metrics store is transient), it should at least not break it. This is addressed in https://github.com/elastic/rally/pull/639/commits/db948833057b467ccf61c05c93d2e9ce26c2f784. @dliappis can you please have another look at those changes?

Patch:
diff --git a/docs/telemetry.rst b/docs/telemetry.rst
index 2711f6d7..1e29c268 100644
--- a/docs/telemetry.rst
+++ b/docs/telemetry.rst
@@ -16,13 +16,14 @@ You probably want to gain additional insights from a race. Therefore, we have ad
 
    Available telemetry devices:
 
-   Command     Name                   Description
-   ----------  ---------------------  -----------------------------------------------------
-   jit         JIT Compiler Profiler  Enables JIT compiler logs.
-   gc          GC log                 Enables GC logs.
-   jfr         Flight Recorder        Enables Java Flight Recorder (requires an Oracle JDK)
-   perf        perf stat              Reads CPU PMU counters (requires Linux and perf)
-   node-stats  Node Stats             Regularly samples node stats
+   Command         Name                   Description
+   --------------  ---------------------  --------------------------------------------------------------------
+   jit             JIT Compiler Profiler  Enables JIT compiler logs.
+   gc              GC log                 Enables GC logs.
+   jfr             Flight Recorder        Enables Java Flight Recorder (requires an Oracle JDK or OpenJDK 11+)
+   perf            perf stat              Reads CPU PMU counters (requires Linux and perf)
+   node-stats      Node Stats             Regularly samples node stats
+   recovery-stats  Recovery Stats         Regularly samples shard recovery stats
 
    Keep in mind that each telemetry device may incur a runtime overhead which can skew results.
 
@@ -78,7 +79,7 @@ node-stats
     Using this telemetry device will skew your results because the node-stats API triggers additional refreshes.
     Additionally a lot of metrics get recorded impacting the measurement results even further.
 
-The node-stats telemetry devices regularly calls the `cluster node-stats API <https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-nodes-stats.html>`_ and records metrics from the following sections:
+The node-stats telemetry device regularly calls the `cluster node-stats API <https://www.elastic.co/guide/en/elasticsearch/reference/current/cluster-nodes-stats.html>`_ and records metrics from the following sections:
 
 * Indices stats (key ``indices`` in the node-stats API)
 * Thread pool stats (key ``jvm.thread_pool`` in the node-stats API)
@@ -100,3 +101,13 @@ Supported telemetry parameters:
 * ``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.
+
+recovery-stats
+--------------
+
+The recovery-stats telemetry device regularly calls the `indices recovery API <https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-recovery.html>`_ and records one metrics document per shard.
+
+Supported telemetry parameters:
+
+* ``recovery-stats-indices`` (default: all indices): An index pattern for which recovery stats should be checked.
+* ``recovery-stats-sample-interval`` (default 1): A positive number greater than zero denoting the sampling interval in seconds.
diff --git a/esrally/mechanic/launcher.py b/esrally/mechanic/launcher.py
index cd345fbc..80a9cc97 100644
--- a/esrally/mechanic/launcher.py
+++ b/esrally/mechanic/launcher.py
@@ -94,7 +94,8 @@ class ClusterLauncher:
             telemetry.GcTimesSummary(es_default, self.metrics_store),
             telemetry.IndexStats(es_default, self.metrics_store),
             telemetry.MlBucketProcessingTime(es_default, self.metrics_store),
-            telemetry.CcrStats(telemetry_params, es, self.metrics_store)
+            telemetry.CcrStats(telemetry_params, es, self.metrics_store),
+            telemetry.RecoveryStats(telemetry_params, es, self.metrics_store)
         ])
 
         # The list of nodes will be populated by ClusterMetaDataInfo, so no need to do it here
diff --git a/esrally/mechanic/telemetry.py b/esrally/mechanic/telemetry.py
index 9f2fadff..74f9cf10 100644
--- a/esrally/mechanic/telemetry.py
+++ b/esrally/mechanic/telemetry.py
@@ -31,7 +31,7 @@ from esrally.metrics import MetaInfoScope
 
 def list_telemetry():
     console.println("Available telemetry devices:\n")
-    devices = [[device.command, device.human_name, device.help] for device in [JitCompiler, Gc, FlightRecorder, PerfStat, NodeStats]]
+    devices = [[device.command, device.human_name, device.help] for device in [JitCompiler, Gc, FlightRecorder, PerfStat, NodeStats, RecoveryStats]]
     console.println(tabulate.tabulate(devices, ["Command", "Name", "Description"]))
     console.println("\nKeep in mind that each telemetry device may incur a runtime overhead which can skew results.")
 
@@ -441,6 +441,127 @@ class CcrStatsRecorder:
                 self.metrics_store.put_doc(shard_stats, level=MetaInfoScope.cluster, meta_data=shard_metadata)
 
 
+class RecoveryStats(TelemetryDevice):
+    internal = False
+    command = "recovery-stats"
+    human_name = "Recovery Stats"
+    help = "Regularly samples shard recovery stats"
+
+    """
+    Gathers recovery stats on a cluster level
+    """
+
+    def __init__(self, telemetry_params, clients, metrics_store):
+        """
+        :param telemetry_params: The configuration object for telemetry_params.
+            May optionally specify:
+            ``recovery-stats-indices``: JSON structure specifying the index pattern per cluster to publish stats from.
+            Not all clusters need to be specified, but any name used must be be present in target.hosts. Alternatively,
+            the index pattern can be specified as a string can be specified in case only one cluster is involved.
+            Example:
+            {"recovery-stats-indices": {"cluster_a": ["follower"],"default": ["leader"]}
+
+            ``recovery-stats-sample-interval``: positive integer controlling the sampling interval. Default: 1 second.
+        :param clients: A dict of clients to all clusters.
+        :param metrics_store: The configured metrics store we write to.
+        """
+        super().__init__()
+
+        self.telemetry_params = telemetry_params
+        self.clients = clients
+        self.sample_interval = telemetry_params.get("recovery-stats-sample-interval", 1)
+        if self.sample_interval <= 0:
+            raise exceptions.SystemSetupError(
+                "The telemetry parameter 'recovery-stats-sample-interval' must be greater than zero but was {}."
+                    .format(self.sample_interval))
+        self.specified_cluster_names = self.clients.keys()
+        indices_per_cluster = self.telemetry_params.get("recovery-stats-indices", False)
+        # allow the user to specify either an index pattern as string or as a JSON object
+        if isinstance(indices_per_cluster, str):
+            self.indices_per_cluster = {opts.TargetHosts.DEFAULT: indices_per_cluster}
+        else:
+            self.indices_per_cluster = indices_per_cluster
+
+        if self.indices_per_cluster:
+            for cluster_name in self.indices_per_cluster.keys():
+                if cluster_name not in clients:
+                    raise exceptions.SystemSetupError(
+                        "The telemetry parameter 'recovery-stats-indices' must be a JSON Object with keys matching "
+                        "the cluster names [{}] specified in --target-hosts "
+                        "but it had [{}].".format(",".join(sorted(clients.keys())), cluster_name))
+            self.specified_cluster_names = self.indices_per_cluster.keys()
+
+        self.metrics_store = metrics_store
+        self.samplers = []
+
+    def on_benchmark_start(self):
+        for cluster_name in self.specified_cluster_names:
+            recorder = RecoveryStatsRecorder(cluster_name, self.clients[cluster_name], self.metrics_store,
+                                             self.sample_interval,
+                                             self.indices_per_cluster[cluster_name] if self.indices_per_cluster else "")
+            sampler = SamplerThread(recorder)
+            self.samplers.append(sampler)
+            sampler.setDaemon(True)
+            # we don't require starting recorders precisely at the same time
+            sampler.start()
+
+    def on_benchmark_stop(self):
+        if self.samplers:
+            for sampler in self.samplers:
+                sampler.finish()
+
+
+class RecoveryStatsRecorder:
+    """
+    Collects and pushes recovery stats for the specified cluster to the metric store.
+    """
+
+    def __init__(self, cluster_name, client, metrics_store, sample_interval, indices=None):
+        """
+        :param cluster_name: The cluster_name that the client connects to, as specified in target.hosts.
+        :param client: The Elasticsearch client for this cluster.
+        :param metrics_store: The configured metrics store we write to.
+        :param sample_interval: integer controlling the interval, in seconds, between collecting samples.
+        :param indices: optional list of indices to filter results from.
+        """
+
+        self.cluster_name = cluster_name
+        self.client = client
+        self.metrics_store = metrics_store
+        self.sample_interval = sample_interval
+        self.indices = indices
+        self.logger = logging.getLogger(__name__)
+
+    def __str__(self):
+        return "recovery stats"
+
+    def record(self):
+        """
+        Collect recovery stats for indexes (optionally) specified in telemetry parameters and push to metrics store.
+        """
+        import elasticsearch
+
+        try:
+            stats = self.client.indices.recovery(index=self.indices, active_only=True, detailed=False)
+        except elasticsearch.TransportError:
+            msg = "A transport error occurred while collecting recovery stats on cluster [{}]".format(self.cluster_name)
+            self.logger.exception(msg)
+            raise exceptions.RallyError(msg)
+
+        for idx, idx_stats in stats.items():
+            for shard in idx_stats["shards"]:
+                doc = {
+                    "name": "recovery-stats",
+                    "shard": shard
+                }
+                shard_metadata = {
+                    "cluster": self.cluster_name,
+                    "index": idx,
+                    "shard": shard["id"]
+                }
+                self.metrics_store.put_doc(doc, level=MetaInfoScope.cluster, meta_data=shard_metadata)
+
+
 class NodeStats(TelemetryDevice):
     """
     Gathers different node stats.
diff --git a/esrally/utils/opts.py b/esrally/utils/opts.py
index 7b3bc45c..a44940ab 100644
--- a/esrally/utils/opts.py
+++ b/esrally/utils/opts.py
@@ -102,6 +102,8 @@ class ConnectOptions:
 
 
 class TargetHosts(ConnectOptions):
+    DEFAULT = "default"
+
     def __init__(self, argvalue):
         self.argname = "--target-hosts"
         self.argvalue = argvalue
@@ -118,7 +120,7 @@ class TargetHosts(ConnectOptions):
             defined as a json string or file.
             """
 
-            return {"default": _normalize_hosts(arg)}
+            return {TargetHosts.DEFAULT: _normalize_hosts(arg)}
 
         self.parsed_options = to_dict(self.argvalue, default_parser=normalize_to_dict)
 
@@ -154,7 +156,7 @@ class ClientOptions(ConnectOptions):
             defined as a json string or file.
             """
 
-            return {"default": kv_to_map(arg)}
+            return {TargetHosts.DEFAULT: kv_to_map(arg)}
 
         if self.argvalue == ClientOptions.DEFAULT_CLIENT_OPTIONS and self.target_hosts != None:
             # --client-options unset but multi-clusters used in --target-hosts? apply options defaults for all cluster names.


Test patch:
diff --git a/tests/mechanic/telemetry_test.py b/tests/mechanic/telemetry_test.py
index 7134cca8..cbef046f 100644
--- a/tests/mechanic/telemetry_test.py
+++ b/tests/mechanic/telemetry_test.py
@@ -157,9 +157,10 @@ class Client:
 
 
 class SubClient:
-    def __init__(self, stats=None, info=None):
+    def __init__(self, stats=None, info=None, recovery=None):
         self._stats = stats
         self._info = info
+        self._recovery = recovery
 
     def stats(self, *args, **kwargs):
         return self._stats
@@ -167,6 +168,9 @@ class SubClient:
     def info(self, *args, **kwargs):
         return self._info
 
+    def recovery(self, *args, **kwargs):
+        return self._recovery
+
 
 class TransportClient:
     def __init__(self, response=None, force_error=False):
@@ -625,6 +629,219 @@ class CcrStatsRecorderTests(TestCase):
         )
 
 
+class RecoveryStatsTests(TestCase):
+    @mock.patch("esrally.metrics.EsMetricsStore.put_doc")
+    def test_no_metrics_if_no_pending_recoveries(self, metrics_store_put_doc):
+        response = {}
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        client = Client(indices=SubClient(recovery=response))
+        recorder = telemetry.RecoveryStatsRecorder(cluster_name="leader",
+                                                   client=client,
+                                                   metrics_store=metrics_store,
+                                                   sample_interval=1,
+                                                   indices=["index1"])
+        recorder.record()
+
+        self.assertEqual(0, metrics_store_put_doc.call_count)
+
+    @mock.patch("esrally.metrics.EsMetricsStore.put_doc")
+    def test_stores_single_shard_stats(self, metrics_store_put_doc):
+        response = {
+            "index1": {
+                "shards": [{
+                    "id": 0,
+                    "type": "STORE",
+                    "stage": "DONE",
+                    "primary": True,
+                    "start_time": "2014-02-24T12:38:06.349",
+                    "start_time_in_millis": "1393245486349",
+                    "stop_time": "2014-02-24T12:38:08.464",
+                    "stop_time_in_millis": "1393245488464",
+                    "total_time": "2.1s",
+                    "total_time_in_millis": 2115,
+                    "source": {
+                        "id": "RGMdRc-yQWWKIBM4DGvwqQ",
+                        "host": "my.fqdn",
+                        "transport_address": "my.fqdn",
+                        "ip": "10.0.1.7",
+                        "name": "my_es_node"
+                    },
+                    "target": {
+                        "id": "RGMdRc-yQWWKIBM4DGvwqQ",
+                        "host": "my.fqdn",
+                        "transport_address": "my.fqdn",
+                        "ip": "10.0.1.7",
+                        "name": "my_es_node"
+                    },
+                    "index": {
+                        "size": {
+                            "total": "24.7mb",
+                            "total_in_bytes": 26001617,
+                            "reused": "24.7mb",
+                            "reused_in_bytes": 26001617,
+                            "recovered": "0b",
+                            "recovered_in_bytes": 0,
+                            "percent": "100.0%"
+                        },
+                        "files": {
+                            "total": 26,
+                            "reused": 26,
+                            "recovered": 0,
+                            "percent": "100.0%"
+                        },
+                        "total_time": "2ms",
+                        "total_time_in_millis": 2,
+                        "source_throttle_time": "0s",
+                        "source_throttle_time_in_millis": 0,
+                        "target_throttle_time": "0s",
+                        "target_throttle_time_in_millis": 0
+                    },
+                    "translog": {
+                        "recovered": 71,
+                        "total": 0,
+                        "percent": "100.0%",
+                        "total_on_start": 0,
+                        "total_time": "2.0s",
+                        "total_time_in_millis": 2025
+                    },
+                    "verify_index": {
+                        "check_index_time": 0,
+                        "check_index_time_in_millis": 0,
+                        "total_time": "88ms",
+                        "total_time_in_millis": 88
+                    }
+                }
+                ]
+            }
+        }
+
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        client = Client(indices=SubClient(recovery=response))
+        recorder = telemetry.RecoveryStatsRecorder(cluster_name="leader",
+                                                   client=client,
+                                                   metrics_store=metrics_store,
+                                                   sample_interval=1,
+                                                   indices=["index1"])
+        recorder.record()
+
+        shard_metadata = {
+            "cluster": "leader",
+            "index": "index1",
+            "shard": 0
+        }
+
+        metrics_store_put_doc.assert_has_calls([
+            mock.call({
+                "name": "recovery-stats",
+                "shard": response["index1"]["shards"][0]
+            }, level=MetaInfoScope.cluster, meta_data=shard_metadata)
+        ],  any_order=True)
+
+    @mock.patch("esrally.metrics.EsMetricsStore.put_doc")
+    def test_stores_multi_index_multi_shard_stats(self, metrics_store_put_doc):
+        response = {
+            "index1": {
+                "shards": [
+                    {
+                        # for the test we only assume a subset of the fields
+                        "id": 0,
+                        "type": "STORE",
+                        "stage": "DONE",
+                        "primary": True,
+                        "total_time_in_millis": 100
+                    },
+                    {
+                        "id": 1,
+                        "type": "STORE",
+                        "stage": "DONE",
+                        "primary": True,
+                        "total_time_in_millis": 200
+                    }
+                ]
+            },
+            "index2": {
+                "shards": [
+                    {
+                        "id": 0,
+                        "type": "STORE",
+                        "stage": "DONE",
+                        "primary": True,
+                        "total_time_in_millis": 300
+                    },
+                    {
+                        "id": 1,
+                        "type": "STORE",
+                        "stage": "DONE",
+                        "primary": True,
+                        "total_time_in_millis": 400
+                    },
+                    {
+                        "id": 2,
+                        "type": "STORE",
+                        "stage": "DONE",
+                        "primary": True,
+                        "total_time_in_millis": 500
+                    }
+                ]
+            }
+        }
+
+        cfg = create_config()
+        metrics_store = metrics.EsMetricsStore(cfg)
+        client = Client(indices=SubClient(recovery=response))
+        recorder = telemetry.RecoveryStatsRecorder(cluster_name="leader",
+                                                   client=client,
+                                                   metrics_store=metrics_store,
+                                                   sample_interval=1,
+                                                   indices=["index1", "index2"])
+        recorder.record()
+
+        metrics_store_put_doc.assert_has_calls([
+            mock.call({
+                "name": "recovery-stats",
+                "shard": response["index1"]["shards"][0]
+            }, level=MetaInfoScope.cluster, meta_data={
+                "cluster": "leader",
+                "index": "index1",
+                "shard": 0
+            }),
+            mock.call({
+                "name": "recovery-stats",
+                "shard": response["index1"]["shards"][1]
+            }, level=MetaInfoScope.cluster, meta_data={
+                "cluster": "leader",
+                "index": "index1",
+                "shard": 1
+            }),
+            mock.call({
+                "name": "recovery-stats",
+                "shard": response["index2"]["shards"][0]
+            }, level=MetaInfoScope.cluster, meta_data={
+                "cluster": "leader",
+                "index": "index2",
+                "shard": 0
+            }),
+            mock.call({
+                "name": "recovery-stats",
+                "shard": response["index2"]["shards"][1]
+            }, level=MetaInfoScope.cluster, meta_data={
+                "cluster": "leader",
+                "index": "index2",
+                "shard": 1
+            }),
+            mock.call({
+                "name": "recovery-stats",
+                "shard": response["index2"]["shards"][2]
+            }, level=MetaInfoScope.cluster, meta_data={
+                "cluster": "leader",
+                "index": "index2",
+                "shard": 2
+            }),
+        ],  any_order=True)
+
+
 class NodeStatsTests(TestCase):
     warning = """You have enabled the node-stats telemetry device, but requests to the _nodes/stats Elasticsearch endpoint
           trigger additional refreshes and WILL SKEW results.