Repository: elastic/rally

Issue:
Overhaul confusing index time related metrics
Currently we report various indexing time related metrics like which are based on the indices stats API. However, these numbers are the sum on index level (i.e. sum of all shards). This means that the number is dependent on shard count and will be usually larger than the benchmark's Wall clock time both of which can be confusing to users.

Hence, we should report these numbers on shard level instead and provide min/max values.

Discussion:

Patch:
diff --git a/docs/metrics.rst b/docs/metrics.rst
index 6681f90e..bf24fb91 100644
--- a/docs/metrics.rst
+++ b/docs/metrics.rst
@@ -143,12 +143,12 @@ Rally stores the following metrics:
 * ``segments_terms_memory_in_bytes``: Number of bytes used for terms as reported by the indices stats API.
 * ``segments_norms_memory_in_bytes``: Number of bytes used for norms as reported by the indices stats API.
 * ``segments_points_memory_in_bytes``: Number of bytes used for points as reported by the indices stats API.
-* ``merges_total_time``: Total runtime of merges as reported by the indices stats API. Note that this is not Wall clock time (i.e. if M merge threads ran for N minutes, we will report M * N minutes, not N minutes).
-* ``merges_total_throttled_time``: Total time within merges have been throttled as reported by the indices stats API. Note that this is not Wall clock time.
-* ``indexing_total_time``: Total time used for indexing as reported by the indices stats API. Note that this is not Wall clock time.
-* ``indexing_throttle_time``: Total time that indexing has been throttled as reported by the indices stats API. Note that this is not Wall clock time.
-* ``refresh_total_time``: Total time used for index refresh as reported by the indices stats API. Note that this is not Wall clock time.
-* ``flush_total_time``: Total time used for index flush as reported by the indices stats API. Note that this is not Wall clock time.
+* ``merges_total_time``: Total runtime of merges as reported by the indices stats API. Note that this is not Wall clock time (i.e. if M merge threads ran for N minutes, we will report M * N minutes, not N minutes). These metrics records also have a ``per-shard`` property that contains the times per primary shard in an array.
+* ``merges_total_throttled_time``: Total time within merges have been throttled as reported by the indices stats API. Note that this is not Wall clock time.  These metrics records also have a ``per-shard`` property that contains the times per primary shard in an array.
+* ``indexing_total_time``: Total time used for indexing as reported by the indices stats API. Note that this is not Wall clock time.  These metrics records also have a ``per-shard`` property that contains the times per primary shard in an array.
+* ``indexing_throttle_time``: Total time that indexing has been throttled as reported by the indices stats API. Note that this is not Wall clock time.  These metrics records also have a ``per-shard`` property that contains the times per primary shard in an array.
+* ``refresh_total_time``: Total time used for index refresh as reported by the indices stats API. Note that this is not Wall clock time.  These metrics records also have a ``per-shard`` property that contains the times per primary shard in an array.
+* ``flush_total_time``: Total time used for index flush as reported by the indices stats API. Note that this is not Wall clock time.  These metrics records also have a ``per-shard`` property that contains the times per primary shard in an array.
 * ``final_index_size_bytes``: Final resulting index size on the file system after all nodes have been shutdown at the end of the benchmark. It includes all files in the nodes' data directories (actual index files and translog).
 * ``store_size_in_bytes``: The size in bytes of the index (excluding the translog) as reported by the indices stats API.
 * ``translog_size_in_bytes``: The size in bytes of the translog as reported by the indices stats API.
diff --git a/docs/summary_report.rst b/docs/summary_report.rst
index e883466e..112fdb3d 100644
--- a/docs/summary_report.rst
+++ b/docs/summary_report.rst
@@ -9,36 +9,71 @@ Total indexing time
 * **Definition**: Total time used for indexing as reported by the indices stats API. Note that this is not Wall clock time (i.e. if M indexing threads ran for N minutes, we will report M * N minutes, not N minutes).
 * **Corresponding metrics key**: ``indexing_total_time``
 
+Indexing time per shard
+-----------------------
+
+* **Definition**: Minimum, median and maximum time used for indexing per primary shard as reported by the indices stats API.
+* **Corresponding metrics key**: ``indexing_total_time`` (property: ``per-shard``)
+
 Total indexing throttle time
 ----------------------------
 
 * **Definition**: Total time that indexing has been throttled as reported by the indices stats API. Note that this is not Wall clock time (i.e. if M indexing threads ran for N minutes, we will report M * N minutes, not N minutes).
 * **Corresponding metrics key**: ``indexing_throttle_time``
 
+Indexing throttle time per shard
+--------------------------------
+
+* **Definition**: Minimum, median and maximum time used that indexing has been throttled per primary shard as reported by the indices stats API.
+* **Corresponding metrics key**: ``indexing_throttle_time`` (property: ``per-shard``)
+
 Total merge time
 ----------------
 
 * **Definition**: Total runtime of merges as reported by the indices stats API. Note that this is not Wall clock time.
 * **Corresponding metrics key**: ``merges_total_time``
 
+Merge time per shard
+--------------------
+
+* **Definition**: Minimum, median and maximum time of merges per primary shard as reported by the indices stats API.
+* **Corresponding metrics key**: ``merges_total_time`` (property: ``per-shard``)
+
 Total refresh time
 ------------------
 
 * **Definition**: Total time used for index refresh as reported by the indices stats API. Note that this is not Wall clock time.
 * **Corresponding metrics key**: ``refresh_total_time``
 
+Refresh time per shard
+----------------------
+
+* **Definition**: Minimum, median and maximum time for index refresh per primary shard as reported by the indices stats API.
+* **Corresponding metrics key**: ``refresh_total_time`` (property: ``per-shard``)
+
 Total flush time
 ----------------
 
 * **Definition**: Total time used for index flush as reported by the indices stats API. Note that this is not Wall clock time.
 * **Corresponding metrics key**: ``flush_total_time``
 
+Flush time per shard
+--------------------
+
+* **Definition**: Minimum, median and maximum time for index flush per primary shard as reported by the indices stats API.
+* **Corresponding metrics key**: ``flush_total_time`` (property: ``per-shard``)
+
 Total merge throttle time
 -------------------------
 
 * **Definition**: Total time within merges have been throttled as reported by the indices stats API. Note that this is not Wall clock time.
 * **Corresponding metrics key**: ``merges_total_throttled_time``
 
+Merge throttle time per shard
+-----------------------------
+
+* **Definition**: Minimum, median and maximum time that merges have been throttled per primary shard as reported by the indices stats API.
+* **Corresponding metrics key**: ``merges_total_throttled_time`` (property: ``per-shard``)
 
 Merge time (``X``)
 ------------------
@@ -54,7 +89,7 @@ Where ``X`` is one of:
 
 ..
 
-* **Definition**: Different merge times as reported by Lucene. Only available if Lucene index writer trace logging is enabled (use `--car-params="verbose_iw_logging_enabled:true"` for that).
+* **Definition**: Different merge times as reported by Lucene. Only available if Lucene index writer trace logging is enabled (use ``--car-params="verbose_iw_logging_enabled:true"`` for that).
 * **Corresponding metrics keys**: ``merge_parts_total_time_*``
 
 
diff --git a/esrally/mechanic/telemetry.py b/esrally/mechanic/telemetry.py
index 839b5809..a09ec2c2 100644
--- a/esrally/mechanic/telemetry.py
+++ b/esrally/mechanic/telemetry.py
@@ -7,7 +7,7 @@ import tabulate
 import threading
 
 from esrally import metrics, time, exceptions
-from esrally.utils import io, sysstats, process, console, versions
+from esrally.utils import io, sysstats, console, versions
 
 
 def list_telemetry():
@@ -1047,11 +1047,12 @@ class IndexStats(InternalTelemetryDevice):
         # the pipeline "benchmark-only" where we don't have control over the cluster and the user might not have restarted
         # the cluster so we can at least tell them.
         if self.first_time:
-            index_times = self.index_times(self.index_stats()["primaries"])
-            for k, v in index_times.items():
-                if v > 0:
+            for t in self.index_times(self.index_stats(), per_shard_stats=False):
+                n = t["name"]
+                v = t["value"]
+                if t["value"] > 0:
                     console.warn("%s is %d ms indicating that the cluster is not in a defined clean state. Recorded index time "
-                                 "metrics may be misleading." % (k, v), logger=self.logger)
+                                 "metrics may be misleading." % (n, v), logger=self.logger)
             self.first_time = False
 
     def on_benchmark_stop(self):
@@ -1059,43 +1060,66 @@ class IndexStats(InternalTelemetryDevice):
         self.logger.info("Gathering indices stats for all primaries on benchmark stop.")
         index_stats = self.index_stats()
         self.logger.info("Returned indices stats:\n%s", json.dumps(index_stats, indent=2))
-        if "primaries" not in index_stats:
+        if "_all" not in index_stats or "primaries" not in index_stats["_all"]:
             return
-        p = index_stats["primaries"]
+        p = index_stats["_all"]["primaries"]
         # actually this is add_count
         self.add_metrics(self.extract_value(p, ["segments", "count"]), "segments_count")
         self.add_metrics(self.extract_value(p, ["segments", "memory_in_bytes"]), "segments_memory_in_bytes", "byte")
 
-        for metric_key, value in self.index_times(p).items():
-            self.logger.info("Adding [%s] = [%s] to metrics store.", str(metric_key), str(value))
-            self.add_metrics(value, metric_key, "ms")
+        for t in self.index_times(index_stats):
+            self.metrics_store.put_doc(doc=t, level=metrics.MetaInfoScope.cluster)
 
         self.add_metrics(self.extract_value(p, ["segments", "doc_values_memory_in_bytes"]), "segments_doc_values_memory_in_bytes", "byte")
         self.add_metrics(self.extract_value(p, ["segments", "stored_fields_memory_in_bytes"]), "segments_stored_fields_memory_in_bytes", "byte")
         self.add_metrics(self.extract_value(p, ["segments", "terms_memory_in_bytes"]), "segments_terms_memory_in_bytes", "byte")
         self.add_metrics(self.extract_value(p, ["segments", "norms_memory_in_bytes"]), "segments_norms_memory_in_bytes", "byte")
         self.add_metrics(self.extract_value(p, ["segments", "points_memory_in_bytes"]), "segments_points_memory_in_bytes", "byte")
-        self.add_metrics(self.extract_value(index_stats, ["total", "store", "size_in_bytes"]), "store_size_in_bytes", "byte")
-        self.add_metrics(self.extract_value(index_stats, ["total", "translog", "size_in_bytes"]), "translog_size_in_bytes", "byte")
+        self.add_metrics(self.extract_value(index_stats, ["_all", "total", "store", "size_in_bytes"]), "store_size_in_bytes", "byte")
+        self.add_metrics(self.extract_value(index_stats, ["_all", "total", "translog", "size_in_bytes"]), "translog_size_in_bytes", "byte")
 
     def index_stats(self):
         # noinspection PyBroadException
         try:
-            stats = self.client.indices.stats(metric="_all", level="shards")
-            return stats["_all"]
+            return self.client.indices.stats(metric="_all", level="shards")
         except BaseException:
             self.logger.exception("Could not retrieve index stats.")
             return {}
 
-    def index_times(self, p):
-        return {
-            "merges_total_time": self.extract_value(p, ["merges", "total_time_in_millis"], default_value=0),
-            "merges_total_throttled_time": self.extract_value(p, ["merges", "total_throttled_time_in_millis"], default_value=0),
-            "indexing_total_time": self.extract_value(p, ["indexing", "index_time_in_millis"], default_value=0),
-            "indexing_throttle_time": self.extract_value(p, ["indexing", "throttle_time_in_millis"], default_value=0),
-            "refresh_total_time": self.extract_value(p, ["refresh", "total_time_in_millis"], default_value=0),
-            "flush_total_time": self.extract_value(p, ["flush", "total_time_in_millis"], default_value=0)
-        }
+    def index_times(self, stats, per_shard_stats=True):
+        times = []
+        self.index_time(times, stats, "merges_total_time", ["merges", "total_time_in_millis"], per_shard_stats),
+        self.index_time(times, stats, "merges_total_throttled_time", ["merges", "total_throttled_time_in_millis"], per_shard_stats),
+        self.index_time(times, stats, "indexing_total_time", ["indexing", "index_time_in_millis"], per_shard_stats),
+        self.index_time(times, stats, "indexing_throttle_time", ["indexing", "throttle_time_in_millis"], per_shard_stats),
+        self.index_time(times, stats, "refresh_total_time", ["refresh", "total_time_in_millis"], per_shard_stats),
+        self.index_time(times, stats, "flush_total_time", ["flush", "total_time_in_millis"], per_shard_stats),
+        return times
+
+    def index_time(self, values, stats, name, path, per_shard_stats):
+        primary_total_stats = self.extract_value(stats, ["_all", "primaries"], default_value={})
+        value = self.extract_value(primary_total_stats, path)
+        if value:
+            doc = {
+                "name": name,
+                "value": value,
+                "unit": "ms",
+            }
+            if per_shard_stats:
+                doc["per-shard"] = self.primary_shard_stats(stats, path)
+            values.append(doc)
+
+    def primary_shard_stats(self, stats, path):
+        shard_stats = []
+        try:
+            for idx, shards in stats["indices"].items():
+                for shard_number, shard in shards["shards"].items():
+                    for shard_metrics in shard:
+                        if shard_metrics["routing"]["primary"]:
+                            shard_stats.append(self.extract_value(shard_metrics, path, default_value=0))
+        except KeyError:
+            self.logger.warning("Could not determine primary shard stats at path [%s].", ",".join(path))
+        return shard_stats
 
     def add_metrics(self, value, metric_key, unit=None):
         if value is not None:
diff --git a/esrally/reporter.py b/esrally/reporter.py
index 33204fe9..ea013fd3 100644
--- a/esrally/reporter.py
+++ b/esrally/reporter.py
@@ -3,6 +3,7 @@ import csv
 import io
 import sys
 import logging
+import statistics
 
 import tabulate
 from esrally import metrics, exceptions
@@ -132,11 +133,17 @@ class StatsCalculator:
 
         self.logger.debug("Gathering indexing metrics.")
         result.total_time = self.sum("indexing_total_time")
+        result.total_time_per_shard = self.shard_stats("indexing_total_time")
         result.indexing_throttle_time = self.sum("indexing_throttle_time")
+        result.indexing_throttle_time_per_shard = self.shard_stats("indexing_throttle_time")
         result.merge_time = self.sum("merges_total_time")
+        result.merge_time_per_shard = self.shard_stats("merges_total_time")
         result.refresh_time = self.sum("refresh_total_time")
+        result.refresh_time_per_shard = self.shard_stats("refresh_total_time")
         result.flush_time = self.sum("flush_total_time")
+        result.flush_time_per_shard = self.shard_stats("flush_total_time")
         result.merge_throttle_time = self.sum("merges_total_throttled_time")
+        result.merge_throttle_time_per_shard = self.shard_stats("merges_total_throttled_time")
 
         self.logger.debug("Gathering merge part metrics.")
         result.merge_part_time_postings = self.sum("merge_parts_total_time_postings")
@@ -207,6 +214,25 @@ class StatsCalculator:
                 "unit": unit
             }
 
+    def shard_stats(self, metric_name):
+        values = self.store.get_raw(metric_name, lap=self.lap, mapper=lambda doc: doc["per-shard"])
+        unit = self.store.get_unit(metric_name)
+        if values:
+            flat_values = [w for v in values for w in v]
+            return {
+                "min": min(flat_values),
+                "median": statistics.median(flat_values),
+                "max": max(flat_values),
+                "unit": unit
+            }
+        else:
+            return {
+                "min": None,
+                "median": None,
+                "max": None,
+                "unit": unit
+            }
+
     def error_rate(self, task_name):
         return self.store.get_error_rate(task=task_name, sample_type=metrics.SampleType.Normal, lap=self.lap)
 
@@ -237,11 +263,17 @@ class Stats:
         self.op_metrics = self.v(d, "op_metrics", default=[])
         self.node_metrics = self.v(d, "node_metrics", default=[])
         self.total_time = self.v(d, "total_time")
+        self.total_time_per_shard = self.v(d, "total_time_per_shard")
         self.indexing_throttle_time = self.v(d, "indexing_throttle_time")
+        self.indexing_throttle_time_per_shard = self.v(d, "indexing_throttle_time_per_shard")
         self.merge_time = self.v(d, "merge_time")
+        self.merge_time_per_shard = self.v(d, "merge_time_per_shard")
         self.refresh_time = self.v(d, "refresh_time")
+        self.refresh_time_per_shard = self.v(d, "refresh_time_per_shard")
         self.flush_time = self.v(d, "flush_time")
+        self.flush_time_per_shard = self.v(d, "flush_time_per_shard")
         self.merge_throttle_time = self.v(d, "merge_throttle_time")
+        self.merge_throttle_time_per_shard = self.v(d, "merge_throttle_time_per_shard")
         self.ml_max_processing_time = self.v(d, "ml_max_processing_time")
 
         self.merge_part_time_postings = self.v(d, "merge_part_time_postings")
@@ -459,14 +491,22 @@ class SummaryReporter:
         )
 
     def report_total_times(self, stats):
+        totals = []
+        totals += self.report_total_time("indexing time", stats.total_time, stats.total_time_per_shard)
+        totals += self.report_total_time("indexing throttle time", stats.indexing_throttle_time, stats.indexing_throttle_time_per_shard)
+        totals += self.report_total_time("merge time", stats.merge_time, stats.merge_time_per_shard)
+        totals += self.report_total_time("merge throttle time", stats.merge_throttle_time, stats.merge_throttle_time_per_shard)
+        totals += self.report_total_time("refresh time", stats.refresh_time, stats.refresh_time_per_shard)
+        totals += self.report_total_time("flush time", stats.flush_time, stats.flush_time_per_shard)
+        return totals
+
+    def report_total_time(self, name, total_time, total_time_per_shard):
         unit = "min"
         return self.join(
-            self.line("Total indexing time", "", stats.total_time, unit, convert.ms_to_minutes),
-            self.line("Total indexing throttle time", "", stats.indexing_throttle_time, unit, convert.ms_to_minutes),
-            self.line("Total merge time", "", stats.merge_time, unit, convert.ms_to_minutes),
-            self.line("Total refresh time", "", stats.refresh_time, unit, convert.ms_to_minutes),
-            self.line("Total flush time", "", stats.flush_time, unit, convert.ms_to_minutes),
-            self.line("Total merge throttle time", "", stats.merge_throttle_time, unit, convert.ms_to_minutes)
+            self.line("Total {}".format(name), "", total_time, unit, convert.ms_to_minutes),
+            self.line("Min {} per shard".format(name), "", total_time_per_shard["min"], unit, convert.ms_to_minutes),
+            self.line("Median {} per shard".format(name), "", total_time_per_shard["median"], unit, convert.ms_to_minutes),
+            self.line("Max {} per shard".format(name), "", total_time_per_shard["max"], unit, convert.ms_to_minutes),
         )
 
     def report_merge_part_times(self, stats):
@@ -665,19 +705,38 @@ class ComparisonReporter:
         )
 
     def report_total_times(self, baseline_stats, contender_stats):
+        totals = []
+        totals += self.report_total_time("indexing time",
+                                         baseline_stats.total_time, baseline_stats.total_time_per_shard,
+                                         contender_stats.total_time, contender_stats.total_time_per_shard)
+        totals += self.report_total_time("indexing throttle time",
+                                         baseline_stats.indexing_throttle_time, baseline_stats.indexing_throttle_time_per_shard,
+                                         contender_stats.indexing_throttle_time, contender_stats.indexing_throttle_time_per_shard)
+        totals += self.report_total_time("merge time",
+                                         baseline_stats.merge_time, baseline_stats.merge_time_per_shard,
+                                         contender_stats.merge_time, contender_stats.merge_time_per_shard)
+        totals += self.report_total_time("merge throttle time",
+                                         baseline_stats.merge_throttle_time, baseline_stats.merge_throttle_time_per_shard,
+                                         contender_stats.merge_throttle_time, contender_stats.merge_throttle_time_per_shard)
+        totals += self.report_total_time("refresh time",
+                                         baseline_stats.refresh_time, baseline_stats.refresh_time_per_shard,
+                                         contender_stats.refresh_time, contender_stats.refresh_time_per_shard)
+        totals += self.report_total_time("flush time",
+                                         baseline_stats.flush_time, baseline_stats.flush_time_per_shard,
+                                         contender_stats.flush_time, contender_stats.flush_time_per_shard)
+        return totals
+
+    def report_total_time(self, name, baseline_total, baseline_per_shard, contender_total, contender_per_shard):
+        unit = "min"
         return self.join(
-            self.line("Total indexing time", baseline_stats.total_time, contender_stats.total_time, "", "min",
-                      treat_increase_as_improvement=False, formatter=convert.ms_to_minutes),
-            self.line("Total indexing throttle time", baseline_stats.indexing_throttle_time, contender_stats.indexing_throttle_time, "", "min",
+            self.line("Total {}".format(name), baseline_total, contender_total, "", unit,
                       treat_increase_as_improvement=False, formatter=convert.ms_to_minutes),
-            self.line("Total merge time", baseline_stats.merge_time, contender_stats.merge_time, "", "min",
+            self.line("Min {} per shard".format(name), baseline_per_shard["min"], contender_per_shard["min"], "", unit,
                       treat_increase_as_improvement=False, formatter=convert.ms_to_minutes),
-            self.line("Total refresh time", baseline_stats.refresh_time, contender_stats.refresh_time, "", "min",
+            self.line("Median {} per shard".format(name), baseline_per_shard["median"], contender_per_shard["median"], "", unit,
                       treat_increase_as_improvement=False, formatter=convert.ms_to_minutes),
-            self.line("Total flush time", baseline_stats.flush_time, contender_stats.flush_time, "", "min",
+            self.line("Max {} per shard".format(name), baseline_per_shard["max"], contender_per_shard["max"], "", unit,
                       treat_increase_as_improvement=False, formatter=convert.ms_to_minutes),
-            self.line("Total merge throttle time", baseline_stats.merge_throttle_time, contender_stats.merge_throttle_time, "", "min",
-                      treat_increase_as_improvement=False, formatter=convert.ms_to_minutes)
         )
 
     def report_gc_times(self, baseline_stats, contender_stats):


Test patch:
diff --git a/tests/mechanic/telemetry_test.py b/tests/mechanic/telemetry_test.py
index 40a2ab4c..bd3ae5d1 100644
--- a/tests/mechanic/telemetry_test.py
+++ b/tests/mechanic/telemetry_test.py
@@ -1385,9 +1385,10 @@ class GcTimesSummaryTests(TestCase):
 
 
 class IndexStatsTests(TestCase):
+    @mock.patch("esrally.metrics.EsMetricsStore.put_doc")
     @mock.patch("esrally.metrics.EsMetricsStore.put_value_cluster_level")
     @mock.patch("esrally.metrics.EsMetricsStore.put_count_cluster_level")
-    def test_stores_available_index_stats(self, metrics_store_cluster_count, metrics_store_cluster_value):
+    def test_stores_available_index_stats(self, metrics_store_cluster_count, metrics_store_cluster_value, metrics_store_put_doc):
         client = Client(indices=SubClient({
             "_all": {
                 "primaries": {
@@ -1429,17 +1430,17 @@ class IndexStatsTests(TestCase):
                         "points_memory_in_bytes": 512
                     },
                     "merges": {
-                        "total_time_in_millis": 300,
-                        "total_throttled_time_in_millis": 120
+                        "total_time_in_millis": 509341,
+                        "total_throttled_time_in_millis": 98925
                     },
                     "indexing": {
-                        "index_time_in_millis": 2000
+                        "index_time_in_millis": 1065688
                     },
                     "refresh": {
-                        "total_time_in_millis": 200
+                        "total_time_in_millis": 158465
                     },
                     "flush": {
-                        "total_time_in_millis": 100
+                        "total_time_in_millis": 19082
                     }
                 },
                 "total": {
@@ -1453,21 +1454,144 @@ class IndexStatsTests(TestCase):
                         "uncommitted_size_in_bytes": 430
                     }
                 }
+            },
+            "indices": {
+                "idx-001": {
+                    "shards": {
+                        "0": [
+                            {
+                                "routing": {
+                                    "primary": False
+                                },
+                                "indexing": {
+                                    "index_total": 2280171,
+                                    "index_time_in_millis": 533662,
+                                    "throttle_time_in_millis": 0
+                                },
+                                "merges": {
+                                    "total_time_in_millis": 280689,
+                                    "total_stopped_time_in_millis": 0,
+                                    "total_throttled_time_in_millis": 58846,
+                                    "total_auto_throttle_in_bytes": 8085428
+                                },
+                                "refresh": {
+                                    "total_time_in_millis": 81004
+                                },
+                                "flush": {
+                                    "total_time_in_millis": 9879
+                                }
+                            }
+                        ],
+                        "1": [
+                            {
+                                "routing": {
+                                    "primary": True,
+                                },
+                                "indexing": {
+                                    "index_time_in_millis": 532026,
+                                },
+                                "merges": {
+                                    "total_time_in_millis": 228652,
+                                    "total_throttled_time_in_millis": 40079,
+                                },
+                                "refresh": {
+                                    "total_time_in_millis": 77461,
+                                },
+                                "flush": {
+                                    "total_time_in_millis": 9203
+                                }
+                            }
+                        ]
+                    }
+                },
+                "idx-002": {
+                    "shards": {
+                        "0": [
+                            {
+                                "routing": {
+                                    "primary": True,
+                                },
+                                "indexing": {
+                                    "index_time_in_millis": 533662,
+                                },
+                                "merges": {
+                                    "total_time_in_millis": 280689,
+                                    "total_throttled_time_in_millis": 58846,
+                                },
+                                "refresh": {
+                                    "total_time_in_millis": 81004,
+                                },
+                                "flush": {
+                                    "total_time_in_millis": 9879
+                                }
+                            }
+                        ],
+                        "1": [
+                            {
+                                "routing": {
+                                    "primary": False,
+                                },
+                                "indexing": {
+                                    "index_time_in_millis": 532026,
+                                    "throttle_time_in_millis": 296
+                                },
+                                "merges": {
+                                    "total_time_in_millis": 228652,
+                                    "total_throttled_time_in_millis": 40079,
+                                },
+                                "refresh": {
+                                    "total_time_in_millis": 77461,
+                                },
+                                "flush": {
+                                    "total_time_in_millis": 9203
+                                }
+                            }
+                        ]
+                    }
+                }
             }
         })
 
         t.on_benchmark_stop()
 
+        metrics_store_put_doc.assert_has_calls([
+            mock.call(doc={
+                "name": "merges_total_time",
+                "value": 509341,
+                "unit": "ms",
+                "per-shard": [228652, 280689]
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "merges_total_throttled_time",
+                "value": 98925,
+                "unit": "ms",
+                "per-shard": [40079, 58846]
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "indexing_total_time",
+                "value": 1065688,
+                "unit": "ms",
+                "per-shard": [532026, 533662]
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "refresh_total_time",
+                "value": 158465,
+                "unit": "ms",
+                "per-shard": [77461, 81004]
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "flush_total_time",
+                "value": 19082,
+                "unit": "ms",
+                "per-shard": [9203, 9879]
+            }, level=metrics.MetaInfoScope.cluster),
+        ])
+
         metrics_store_cluster_count.assert_has_calls([
             mock.call("segments_count", 5)
         ])
         metrics_store_cluster_value.assert_has_calls([
             mock.call("segments_memory_in_bytes", 2048, "byte"),
-            mock.call("merges_total_time", 300, "ms"),
-            mock.call("merges_total_throttled_time", 120, "ms"),
-            mock.call("indexing_total_time", 2000, "ms"),
-            mock.call("refresh_total_time", 200, "ms"),
-            mock.call("flush_total_time", 100, "ms"),
             mock.call("segments_doc_values_memory_in_bytes", 128, "byte"),
             mock.call("segments_stored_fields_memory_in_bytes", 1024, "byte"),
             mock.call("segments_terms_memory_in_bytes", 256, "byte"),
@@ -1476,9 +1600,10 @@ class IndexStatsTests(TestCase):
             mock.call("translog_size_in_bytes", 2647984713, "byte"),
         ], any_order=True)
 
+    @mock.patch("esrally.metrics.EsMetricsStore.put_doc")
     @mock.patch("esrally.metrics.EsMetricsStore.put_value_cluster_level")
     @mock.patch("esrally.metrics.EsMetricsStore.put_count_cluster_level")
-    def test_index_stats_are_per_lap(self, metrics_store_cluster_count, metrics_store_cluster_value):
+    def test_index_stats_are_per_lap(self, metrics_store_cluster_count, metrics_store_cluster_value, metrics_store_put_doc):
         client = Client(indices=SubClient({
             "_all": {
                 "primaries": {
@@ -1569,14 +1694,74 @@ class IndexStatsTests(TestCase):
 
         t.on_benchmark_stop()
 
+        metrics_store_put_doc.assert_has_calls([
+            # 1st lap
+            mock.call(doc={
+                "name": "merges_total_time",
+                "value": 300,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "merges_total_throttled_time",
+                "value": 120,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "indexing_total_time",
+                "value": 2000,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "refresh_total_time",
+                "value": 200,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "flush_total_time",
+                "value": 100,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            # 2nd lap
+            mock.call(doc={
+                "name": "merges_total_time",
+                "value": 900,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "merges_total_throttled_time",
+                "value": 120,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "indexing_total_time",
+                "value": 8000,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "refresh_total_time",
+                "value": 500,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+            mock.call(doc={
+                "name": "flush_total_time",
+                "value": 300,
+                "unit": "ms",
+                "per-shard": []
+            }, level=metrics.MetaInfoScope.cluster),
+        ])
+
         metrics_store_cluster_value.assert_has_calls([
             # 1st lap
             mock.call("segments_memory_in_bytes", 2048, "byte"),
-            mock.call("merges_total_time", 300, "ms"),
-            mock.call("merges_total_throttled_time", 120, "ms"),
-            mock.call("indexing_total_time", 2000, "ms"),
-            mock.call("refresh_total_time", 200, "ms"),
-            mock.call("flush_total_time", 100, "ms"),
             mock.call("segments_doc_values_memory_in_bytes", 128, "byte"),
             mock.call("segments_stored_fields_memory_in_bytes", 1024, "byte"),
             mock.call("segments_terms_memory_in_bytes", 256, "byte"),
@@ -1584,11 +1769,6 @@ class IndexStatsTests(TestCase):
 
             # 2nd lap
             mock.call("segments_memory_in_bytes", 2048, "byte"),
-            mock.call("merges_total_time", 900, "ms"),
-            mock.call("merges_total_throttled_time", 120, "ms"),
-            mock.call("indexing_total_time", 8000, "ms"),
-            mock.call("refresh_total_time", 500, "ms"),
-            mock.call("flush_total_time", 300, "ms"),
             mock.call("segments_doc_values_memory_in_bytes", 128, "byte"),
             mock.call("segments_stored_fields_memory_in_bytes", 1024, "byte"),
             mock.call("segments_terms_memory_in_bytes", 256, "byte"),