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):

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.
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"),

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-benchmark
apt-get update apt-get install -y gcc
3.6
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs==22.2.0
certifi==2021.5.30
elasticsearch==6.2.0
-e git+https://github.com/elastic/rally.git@fc63d066a0a28fe7a683d4d41acc2ab2bd53d522#egg=esrally
importlib-metadata==4.8.3
iniconfig==1.1.1
Jinja2==2.9.5
jsonschema==2.5.1
MarkupSafe==2.0.1
packaging==21.3
pluggy==1.0.0
psutil==5.4.0
py==1.11.0
py-cpuinfo==3.2.0
pyparsing==3.1.4
pytest==7.0.1
pytest-benchmark==3.4.1
tabulate==0.8.1
thespian==3.9.2
tomli==1.2.3
typing_extensions==4.1.1
urllib3==1.22
zipp==3.6.0

name: rally
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - attrs==22.2.0
      - elasticsearch==6.2.0
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - jinja2==2.9.5
      - jsonschema==2.5.1
      - markupsafe==2.0.1
      - packaging==21.3
      - pluggy==1.0.0
      - psutil==5.4.0
      - py==1.11.0
      - py-cpuinfo==3.2.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - pytest-benchmark==3.4.1
      - tabulate==0.8.1
      - thespian==3.9.2
      - tomli==1.2.3
      - typing-extensions==4.1.1
      - urllib3==1.22
      - zipp==3.6.0
prefix: /opt/conda/envs/rally

tests/mechanic/telemetry_test.py::IndexStatsTests::test_index_stats_are_per_lap tests/mechanic/telemetry_test.py::IndexStatsTests::test_stores_available_index_stats
tests/mechanic/telemetry_test.py::TelemetryTests::test_merges_options_set_by_different_devices tests/mechanic/telemetry_test.py::StartupTimeTests::test_store_calculated_metrics tests/mechanic/telemetry_test.py::MergePartsDeviceTests::test_store_calculated_metrics tests/mechanic/telemetry_test.py::MergePartsDeviceTests::test_store_nothing_if_no_metrics_present tests/mechanic/telemetry_test.py::JfrTests::test_sets_options_for_java_9_or_above_custom_recording_template tests/mechanic/telemetry_test.py::JfrTests::test_sets_options_for_java_9_or_above_default_recording_template tests/mechanic/telemetry_test.py::JfrTests::test_sets_options_for_pre_java_9_custom_recording_template tests/mechanic/telemetry_test.py::JfrTests::test_sets_options_for_pre_java_9_default_recording_template tests/mechanic/telemetry_test.py::GcTests::test_sets_options_for_java_9_or_above tests/mechanic/telemetry_test.py::GcTests::test_sets_options_for_pre_java_9 tests/mechanic/telemetry_test.py::CcrStatsTests::test_negative_sample_interval_forbidden tests/mechanic/telemetry_test.py::CcrStatsTests::test_wrong_cluster_name_in_ccr_stats_indices_forbidden tests/mechanic/telemetry_test.py::CcrStatsRecorderTests::test_raises_exception_on_transport_error tests/mechanic/telemetry_test.py::CcrStatsRecorderTests::test_stores_default_ccr_stats tests/mechanic/telemetry_test.py::CcrStatsRecorderTests::test_stores_default_ccr_stats_many_shards tests/mechanic/telemetry_test.py::CcrStatsRecorderTests::test_stores_filtered_ccr_stats tests/mechanic/telemetry_test.py::NodeStatsRecorderTests::test_negative_sample_interval_forbidden tests/mechanic/telemetry_test.py::NodeStatsRecorderTests::test_stores_all_nodes_stats tests/mechanic/telemetry_test.py::NodeStatsRecorderTests::test_stores_default_nodes_stats tests/mechanic/telemetry_test.py::ClusterEnvironmentInfoTests::test_stores_cluster_level_metrics_on_attach tests/mechanic/telemetry_test.py::NodeEnvironmentInfoTests::test_stores_node_level_metrics_on_attach tests/mechanic/telemetry_test.py::ExternalEnvironmentInfoTests::test_fallback_when_host_not_available tests/mechanic/telemetry_test.py::ExternalEnvironmentInfoTests::test_stores_all_node_metrics_on_attach tests/mechanic/telemetry_test.py::ClusterMetaDataInfoTests::test_enriches_cluster_nodes_for_elasticsearch_1_x tests/mechanic/telemetry_test.py::ClusterMetaDataInfoTests::test_enriches_cluster_nodes_for_elasticsearch_after_1_x tests/mechanic/telemetry_test.py::GcTimesSummaryTests::test_stores_only_diff_of_gc_times tests/mechanic/telemetry_test.py::IndexSizeTests::test_stores_index_size_for_data_paths tests/mechanic/telemetry_test.py::IndexSizeTests::test_stores_nothing_if_no_data_path
docs/metrics.rst esrally/mechanic/telemetry.py esrally/reporter.py docs/summary_report.rst
docs/metrics.rst esrally/mechanic/telemetry.py esrally/reporter.py docs/summary_report.rst