Repository: elastic/rally

Issue:
es download url below 6.3.0 version is not correct
- rally version: 1.0.1

when rally download es, it's url is "http://xxxx/elasticsearch-oss-6.2.2.tar.gz", but correct url should be "http://xxxx/elasticsearch-6.2.2.tar.gz".  The former url is not applied to versions below 6.3.0.

Discussion:

Patch:
diff --git a/Dockerfile b/Dockerfile
index 3a91b92c..55ae662e 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -5,14 +5,14 @@ ARG NEW_USER
 
 RUN apt-get -y update && \
     apt-get install -y curl python3 python3-pip vim openssh-client \
-    git make build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev \
+    git make build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev libsqlite3-dev libffi-dev \
     openjdk-8-jdk
 
 RUN pip3 install tox virtualenvwrapper twine sphinx sphinx_rtd_theme wheel
 
-RUN curl --progress-bar -o openjdk-10.0.1_linux-x64_bin.tar.gz \
-    https://download.java.net/java/GA/jdk10/10.0.1/fb4372174a714e6b8c52526dc134031e/10/openjdk-10.0.1_linux-x64_bin.tar.gz && \
-    mkdir -p /opt/jdk-10 ; tar xzf openjdk-10.0.1_linux-x64_bin.tar.gz -C /opt/jdk-10 --strip-components 1 && rm openjdk-10.0.1_linux-x64_bin.tar.gz
+RUN curl --progress-bar -o openjdk-11_linux-x64_bin.tar.gz \
+    https://download.java.net/java/ga/jdk11/openjdk-11_linux-x64_bin.tar.gz && \
+    mkdir -p /opt/jdk-11 ; tar xzf openjdk-11_linux-x64_bin.tar.gz -C /opt/jdk-11 --strip-components 1 && rm openjdk-11_linux-x64_bin.tar.gz
 
 RUN useradd -u ${NEW_USER_UID} -U -d /home/${NEW_USER} -s /bin/bash -m ${NEW_USER}
 
@@ -24,12 +24,13 @@ RUN git clone https://github.com/yyuu/pyenv.git ~/.pyenv
 ENV HOME /home/${NEW_USER}
 ENV PYENV_ROOT=/home/${NEW_USER}/.pyenv
 ENV PATH=$PYENV_ROOT/shims:$PYENV_ROOT/bin:$PATH
-ENV JAVA_HOME=/opt/jdk-10
+ENV JAVA_HOME=/opt/jdk-11
 ENV JAVA8_HOME=/usr/lib/jvm/java-8-openjdk-amd64
-ENV JAVA10_HOME=/opt/jdk-10
+ENV JAVA11_HOME=/opt/jdk-11
 ENV RUNTIME_JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
 
 RUN pyenv install 3.4.8
 RUN pyenv install 3.5.5
 RUN pyenv install 3.6.5
-RUN pyenv global 3.6.5 3.5.5 3.4.8
+RUN pyenv install 3.7.0
+RUN pyenv global 3.7.0 3.6.5 3.5.5 3.4.8
diff --git a/docs/command_line_reference.rst b/docs/command_line_reference.rst
index 8c094984..27842c45 100644
--- a/docs/command_line_reference.rst
+++ b/docs/command_line_reference.rst
@@ -405,8 +405,6 @@ We support the following data types:
 * Numbers: There is nothing special about numbers. Example: ``sniffer_timeout:60``
 * Booleans: Specify either ``true`` or ``false``. Example: ``use_ssl:true``
 
-In addition to the options, supported by the Elasticsearch client, it is also possible to enable HTTP compression by specifying ``compressed:true``
-
 Default value: ``timeout:60``
 
 .. warning::
@@ -416,7 +414,7 @@ Default value: ``timeout:60``
 
 Here are a few common examples:
 
-* Enable HTTP compression: ``--client-options="compressed:true"``
+* Enable HTTP compression: ``--client-options="http_compress:true"``
 * Enable basic authentication: ``--client-options="basic_auth_user:'user',basic_auth_password:'password'"``. Avoid the characters ``'``, ``,`` and ``:`` in user name and password as Rally's parsing of these options is currently really simple and there is no possibility to escape characters.
 
 **TLS/SSL**
diff --git a/esrally/client.py b/esrally/client.py
index ac897974..cc3993cc 100644
--- a/esrally/client.py
+++ b/esrally/client.py
@@ -1,4 +1,3 @@
-import gzip
 import logging
 import certifi
 import urllib3
@@ -90,6 +89,15 @@ class EsClientFactory:
         else:
             self.logger.info("HTTP basic authentication: off")
 
+        if self._is_set(self.client_options, "compressed"):
+            console.warn("You set the deprecated client option 'compressed‘. Please use 'http_compress' instead.", logger=self.logger)
+            self.client_options["http_compress"] = self.client_options.pop("compressed")
+
+        if self._is_set(self.client_options, "http_compress"):
+                self.logger.info("HTTP compression: on")
+        else:
+            self.logger.info("HTTP compression: off")
+
     def _is_set(self, client_opts, k):
         try:
             return client_opts[k]
@@ -97,31 +105,5 @@ class EsClientFactory:
             return False
 
     def create(self):
-        class PoolWrap(object):
-            def __init__(self, pool, compressed=False, **kwargs):
-                self.pool = pool
-                self.compressed = compressed
-
-            def urlopen(self, method, url, body, retries, headers, **kw):
-                if body is not None and self.compressed:
-                    body = gzip.compress(body)
-                return self.pool.urlopen(method, url, body=body, retries=retries, headers=headers, **kw)
-
-            def __getattr__(self, attr_name):
-                return getattr(self.pool, attr_name)
-
         import elasticsearch
-
-        class ConfigurableHttpConnection(elasticsearch.Urllib3HttpConnection):
-            def __init__(self, compressed=False, **kwargs):
-                super(ConfigurableHttpConnection, self).__init__(**kwargs)
-                if compressed:
-                    logging.getLogger(__name__).info("HTTP compression: on")
-                    self.headers.update(urllib3.make_headers(accept_encoding=True))
-                    self.headers.update({"Content-Encoding": "gzip"})
-                else:
-                    logging.getLogger(__name__).info("HTTP compression: off")
-                self.pool = PoolWrap(self.pool, **kwargs)
-
-        return elasticsearch.Elasticsearch(hosts=self.hosts, connection_class=ConfigurableHttpConnection,
-                                           ssl_context=self.ssl_context, **self.client_options)
+        return elasticsearch.Elasticsearch(hosts=self.hosts, ssl_context=self.ssl_context, **self.client_options)
diff --git a/esrally/mechanic/telemetry.py b/esrally/mechanic/telemetry.py
index 9eaba9c7..d832814f 100644
--- a/esrally/mechanic/telemetry.py
+++ b/esrally/mechanic/telemetry.py
@@ -391,22 +391,18 @@ class CcrStatsRecorder:
             self.logger.exception(msg)
             raise exceptions.RallyError(msg)
 
-        if self.indices:
-            # Record metrics only for indices specified with telemetry-params.
-            for index in self.indices:
+        if "indices" in stats:
+            for indices in stats["indices"]:
                 try:
-                    if stats[index]:
-                        self.record_stats_per_index(index, stats[index])
+                    if self.indices and indices["index"] not in self.indices:
+                        # Skip metrics for indices not part of user supplied whitelist (ccr-stats-indices) in telemetry params.
+                        continue
+                    self.record_stats_per_index(indices["index"], indices["shards"])
                 except KeyError:
-                    self.logger.warning("The telemetry parameter 'ccr-stats-indices' specified the index [%s] for cluster [%s] but there "
-                                        "are no stats available. Maybe the index hasn't been created yet, ignoring.",
-                                        index, self.cluster_name)
-        else:
-            # Record metrics for every index returned by the CCR stats API.
-            if stats:
-                for index_name, stats in stats.items():
-                    self.record_stats_per_index(index_name, stats)
-            pass
+                    self.logger.warning(
+                        "The 'indices' key in {0} does not contain an 'index' or 'shards' key "
+                        "Maybe the output format of the {0} endpoint has changed. Skipping.".format(ccr_stats_api_endpoint)
+                    )
         time.sleep(self.sample_interval)
 
     def record_stats_per_index(self, name, stats):
@@ -1166,11 +1162,12 @@ class MlBucketProcessingTime(InternalTelemetryDevice):
             for job in results["aggregations"]["jobs"]["buckets"]:
                 ml_job_stats = collections.OrderedDict()
                 ml_job_stats["name"] = "ml_processing_time"
-                ml_job_stats["job_name"] = job["key"]
-                ml_job_stats["min_millis"] = job["min_pt"]["value"]
-                ml_job_stats["mean_millis"] = job["mean_pt"]["value"]
-                ml_job_stats["median_millis"] = job["median_pt"]["values"]["50.0"]
-                ml_job_stats["max_millis"] = job["max_pt"]["value"]
+                ml_job_stats["job"] = job["key"]
+                ml_job_stats["min"] = job["min_pt"]["value"]
+                ml_job_stats["mean"] = job["mean_pt"]["value"]
+                ml_job_stats["median"] = job["median_pt"]["values"]["50.0"]
+                ml_job_stats["max"] = job["max_pt"]["value"]
+                ml_job_stats["unit"] = "ms"
                 self.metrics_store.put_doc(doc=dict(ml_job_stats), level=MetaInfoScope.cluster)
         except KeyError:
             # no ML running
diff --git a/esrally/reporter.py b/esrally/reporter.py
index 3ea2c31d..71d02b4b 100644
--- a/esrally/reporter.py
+++ b/esrally/reporter.py
@@ -154,7 +154,7 @@ class StatsCalculator:
         result.merge_part_time_points = self.sum("merge_parts_total_time_points")
 
         self.logger.debug("Gathering ML max processing times.")
-        result.ml_processing_time = self.all("ml_processing_time")
+        result.ml_processing_time = self.ml_processing_time_stats()
 
         self.logger.debug("Gathering CPU usage metrics.")
         result.median_cpu_usage = self.median("cpu_utilization_1s", sample_type=metrics.SampleType.Normal)
@@ -195,9 +195,6 @@ class StatsCalculator:
     def one(self, metric_name):
         return self.store.get_one(metric_name, lap=self.lap)
 
-    def all(self, metric_name):
-        return self.store.get_raw(metric_name, lap=self.lap)
-
     def summary_stats(self, metric_name, task_name):
         median = self.store.get_median(metric_name, task=task_name, sample_type=metrics.SampleType.Normal, lap=self.lap)
         unit = self.store.get_unit(metric_name, task=task_name)
@@ -231,6 +228,21 @@ class StatsCalculator:
         else:
             return {}
 
+    def ml_processing_time_stats(self):
+        values = self.store.get_raw("ml_processing_time")
+        result = []
+        if values:
+            for v in values:
+                result.append({
+                    "job": v["job"],
+                    "min": v["min"],
+                    "mean": v["mean"],
+                    "median": v["median"],
+                    "max": v["max"],
+                    "unit": v["unit"]
+                })
+        return result
+
     def error_rate(self, task_name):
         return self.store.get_error_rate(task=task_name, sample_type=metrics.SampleType.Normal, lap=self.lap)
 
@@ -272,7 +284,7 @@ class Stats:
         self.flush_time_per_shard = self.v(d, "flush_time_per_shard", default={})
         self.merge_throttle_time = self.v(d, "merge_throttle_time")
         self.merge_throttle_time_per_shard = self.v(d, "merge_throttle_time_per_shard", default={})
-        self.ml_processing_time = self.v(d, "ml_processing_time")
+        self.ml_processing_time = self.v(d, "ml_processing_time", default=[])
 
         self.merge_part_time_postings = self.v(d, "merge_part_time_postings")
         self.merge_part_time_stored_fields = self.v(d, "merge_part_time_stored_fields")
@@ -321,6 +333,18 @@ class Stats:
                         all_results.append(
                             {"task": item["task"], "operation": item["operation"], "name": "error_rate",
                              "value": {"single": item["error_rate"]}})
+            elif metric == "ml_processing_time":
+                for item in value:
+                    all_results.append({
+                        "job": item["job"],
+                        "name": "ml_processing_time",
+                        "value": {
+                            "min": item["min"],
+                            "mean": item["mean"],
+                            "median": item["median"],
+                            "max": item["max"]
+                        }
+                    })
             elif metric == "node_metrics":
                 for item in value:
                     if "startup_time" in item:
@@ -525,11 +549,12 @@ class SummaryReporter:
     def report_ml_processing_times(self, stats):
         lines = []
         for processing_time in stats.ml_processing_time:
-            job_name = processing_time["job_name"]
-            lines.append(self.line("Min ML processing time", job_name, processing_time["min_millis"], "s", convert.ms_to_seconds)),
-            lines.append(self.line("Mean ML processing time", job_name, processing_time["mean_millis"], "s", convert.ms_to_seconds)),
-            lines.append(self.line("Median ML processing time", job_name, processing_time["median_millis"], "s", convert.ms_to_seconds)),
-            lines.append(self.line("Max ML processing time", job_name, processing_time["max_millis"], "s", convert.ms_to_seconds))
+            job_name = processing_time["job"]
+            unit = processing_time["unit"]
+            lines.append(self.line("Min ML processing time", job_name, processing_time["min"], unit)),
+            lines.append(self.line("Mean ML processing time", job_name, processing_time["mean"], unit)),
+            lines.append(self.line("Median ML processing time", job_name, processing_time["median"], unit)),
+            lines.append(self.line("Max ML processing time", job_name, processing_time["max"], unit))
         return lines
 
     def report_cpu_usage(self, stats):
@@ -714,18 +739,19 @@ class ComparisonReporter:
     def report_ml_processing_times(self, baseline_stats, contender_stats):
         lines = []
         for baseline in baseline_stats.ml_processing_time:
-            job_name = baseline["job_name"]
+            job_name = baseline["job"]
+            unit = baseline["unit"]
             # O(n^2) but we assume here only a *very* limited number of jobs (usually just one)
             for contender in contender_stats.ml_processing_time:
-                if contender["job_name"] == job_name:
-                    lines.append(self.line("Min ML processing time", baseline["min_millis"], contender["min_millis"],
-                                           job_name, "s", treat_increase_as_improvement=False, formatter=convert.ms_to_seconds))
-                    lines.append(self.line("Mean ML processing time", baseline["mean_millis"], contender["mean_millis"],
-                                           job_name, "s", treat_increase_as_improvement=False, formatter=convert.ms_to_seconds))
-                    lines.append(self.line("Median ML processing time", baseline["median_millis"], contender["median_millis"],
-                                           job_name, "s", treat_increase_as_improvement=False, formatter=convert.ms_to_seconds))
-                    lines.append(self.line("Max ML processing time", baseline["max_millis"], contender["max_millis"],
-                                           job_name, "s", treat_increase_as_improvement=False, formatter=convert.ms_to_seconds))
+                if contender["job"] == job_name:
+                    lines.append(self.line("Min ML processing time", baseline["min"], contender["min"],
+                                           job_name, unit, treat_increase_as_improvement=False))
+                    lines.append(self.line("Mean ML processing time", baseline["mean"], contender["mean"],
+                                           job_name, unit, treat_increase_as_improvement=False))
+                    lines.append(self.line("Median ML processing time", baseline["median"], contender["median"],
+                                           job_name, unit, treat_increase_as_improvement=False))
+                    lines.append(self.line("Max ML processing time", baseline["max"], contender["max"],
+                                           job_name, unit, treat_increase_as_improvement=False))
         return lines
 
     def report_total_times(self, baseline_stats, contender_stats):
diff --git a/esrally/resources/metrics-template.json b/esrally/resources/metrics-template.json
index 08bcd1a8..d79fd14f 100644
--- a/esrally/resources/metrics-template.json
+++ b/esrally/resources/metrics-template.json
@@ -66,6 +66,18 @@
         "value": {
           "type": "float"
         },
+        "min": {
+          "type": "float"
+        },
+        "max": {
+          "type": "float"
+        },
+        "mean": {
+          "type": "float"
+        },
+        "median": {
+          "type": "float"
+        },
         "unit": {
           "type": "keyword"
         },
@@ -80,6 +92,9 @@
         },
         "operation-type": {
           "type": "keyword"
+        },
+        "job": {
+          "type": "keyword"
         }
       }
     }
diff --git a/esrally/resources/results-template.json b/esrally/resources/results-template.json
index 29b3bc4b..db07f864 100644
--- a/esrally/resources/results-template.json
+++ b/esrally/resources/results-template.json
@@ -74,6 +74,9 @@
         "operation": {
           "type": "keyword"
         },
+        "job": {
+          "type": "keyword"
+        },
         "name": {
           "type": "keyword"
         },
@@ -86,6 +89,9 @@
             "min": {
               "type": "double"
             },
+            "mean": {
+              "type": "double"
+            },
             "median": {
               "type": "double"
             },
diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index d0130b7f..ee49531c 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -44,7 +44,8 @@ def list_tracks(cfg):
 
     data = []
     for t in available_tracks:
-        line = [t.name, t.description, t.number_of_documents, convert.bytes_to_human_string(t.compressed_size_in_bytes),
+        line = [t.name, t.description, convert.number_to_human_string(t.number_of_documents),
+                convert.bytes_to_human_string(t.compressed_size_in_bytes),
                 convert.bytes_to_human_string(t.uncompressed_size_in_bytes)]
         if not only_auto_generated_challenges:
             line.append(t.default_challenge)
diff --git a/esrally/utils/convert.py b/esrally/utils/convert.py
index 2a978713..bc66e052 100644
--- a/esrally/utils/convert.py
+++ b/esrally/utils/convert.py
@@ -25,6 +25,10 @@ def bytes_to_human_string(b):
     return "%d bytes" % b
 
 
+def number_to_human_string(number):
+    return "{:,}".format(number)
+
+
 def mb_to_bytes(mb):
     return int(mb * 1024 * 1024) if mb else mb
 
diff --git a/esrally/utils/repo.py b/esrally/utils/repo.py
index 511fd96d..dd12e97c 100644
--- a/esrally/utils/repo.py
+++ b/esrally/utils/repo.py
@@ -35,7 +35,7 @@ class RallyRepository:
 
     def update(self, distribution_version):
         try:
-            if self.remote and not self.offline:
+            if self.remote:
                 branch = versions.best_match(git.branches(self.repo_dir, remote=self.remote), distribution_version)
                 if branch:
                     # Allow uncommitted changes iff we do not have to change the branch


Test patch:
diff --git a/tests/mechanic/telemetry_test.py b/tests/mechanic/telemetry_test.py
index ba1f5813..d782ca88 100644
--- a/tests/mechanic/telemetry_test.py
+++ b/tests/mechanic/telemetry_test.py
@@ -279,48 +279,59 @@ class CcrStatsRecorderTests(TestCase):
     def test_stores_default_ccr_stats(self, metrics_store_put_doc):
         java_signed_maxlong = CcrStatsRecorderTests.java_signed_maxlong
 
-        shard_id = random.randint(0, 999),
-        leader_global_checkpoint = random.randint(0, java_signed_maxlong),
-        leader_max_seq_no = random.randint(0, java_signed_maxlong),
-        follower_global_checkpoint = random.randint(0, java_signed_maxlong),
-        follower_max_seq_no = random.randint(0, java_signed_maxlong),
-        last_requested_seq_no = random.randint(0, java_signed_maxlong),
-        number_of_concurrent_reads = random.randint(0, java_signed_maxlong),
-        number_of_concurrent_writes = random.randint(0, java_signed_maxlong),
-        number_of_queued_writes = random.randint(0, java_signed_maxlong),
-        index_metadata_version = random.randint(0, java_signed_maxlong),
-        total_fetch_time_millis = random.randint(0, java_signed_maxlong),
-        number_of_successful_fetches = random.randint(0, java_signed_maxlong),
-        number_of_failed_fetches = random.randint(0, java_signed_maxlong),
-        operations_received = random.randint(0, java_signed_maxlong),
-        total_transferred_bytes = random.randint(0, java_signed_maxlong),
-        total_index_time_millis = random.randint(0, java_signed_maxlong),
-        number_of_successful_bulk_operations = random.randint(0, java_signed_maxlong),
-        number_of_failed_bulk_operations = random.randint(0, java_signed_maxlong),
+        shard_id = random.randint(0, 999)
+        leader_index = "leader_cluster:leader"
+        follower_index = "follower"
+        leader_global_checkpoint = random.randint(0, java_signed_maxlong)
+        leader_max_seq_no = random.randint(0, java_signed_maxlong)
+        follower_global_checkpoint = random.randint(0, java_signed_maxlong)
+        follower_max_seq_no = random.randint(0, java_signed_maxlong)
+        last_requested_seq_no = random.randint(0, java_signed_maxlong)
+        number_of_concurrent_reads = random.randint(0, java_signed_maxlong)
+        number_of_concurrent_writes = random.randint(0, java_signed_maxlong)
+        number_of_queued_writes = random.randint(0, java_signed_maxlong)
+        mapping_version = random.randint(0, java_signed_maxlong)
+        total_fetch_time_millis = random.randint(0, java_signed_maxlong)
+        number_of_successful_fetches = random.randint(0, java_signed_maxlong)
+        number_of_failed_fetches = random.randint(0, java_signed_maxlong)
+        operations_received = random.randint(0, java_signed_maxlong)
+        total_transferred_bytes = random.randint(0, java_signed_maxlong)
+        total_index_time_millis = random.randint(0, java_signed_maxlong)
+        time_since_last_fetch_millis = random.randint(0, java_signed_maxlong)
+        number_of_successful_bulk_operations = random.randint(0, java_signed_maxlong)
+        number_of_failed_bulk_operations = random.randint(0, java_signed_maxlong)
         number_of_operations_indexed = random.randint(0, java_signed_maxlong)
 
         ccr_stats_follower_response = {
-            "follower": [
+            "indices": [
                 {
-                    "shard_id": shard_id,
-                    "follower_global_checkpoint": follower_global_checkpoint,
-                    "follower_max_seq_no": follower_max_seq_no,
-                    "index_metadata_version": index_metadata_version,
-                    "last_requested_seq_no": last_requested_seq_no,
-                    "leader_global_checkpoint": leader_global_checkpoint,
-                    "leader_max_seq_no": leader_max_seq_no,
-                    "number_of_concurrent_reads": number_of_concurrent_reads,
-                    "number_of_concurrent_writes": number_of_concurrent_writes,
-                    "number_of_failed_bulk_operations": number_of_failed_bulk_operations,
-                    "number_of_failed_fetches": number_of_failed_fetches,
-                    "number_of_operations_indexed": number_of_operations_indexed,
-                    "number_of_queued_writes": number_of_queued_writes,
-                    "number_of_successful_bulk_operations": number_of_successful_bulk_operations,
-                    "number_of_successful_fetches": number_of_successful_fetches,
-                    "operations_received": operations_received,
-                    "total_fetch_time_millis": total_fetch_time_millis,
-                    "total_index_time_millis": total_index_time_millis,
-                    "total_transferred_bytes": total_transferred_bytes
+                    "index": follower_index,
+                    "shards": [
+                        {
+                            "shard_id": shard_id,
+                            "follower_global_checkpoint": follower_global_checkpoint,
+                            "leader_index": leader_index,
+                            "follower_index": follower_index,
+                            "follower_max_seq_no": follower_max_seq_no,
+                            "mapping_version": mapping_version,
+                            "last_requested_seq_no": last_requested_seq_no,
+                            "leader_global_checkpoint": leader_global_checkpoint,
+                            "leader_max_seq_no": leader_max_seq_no,
+                            "number_of_concurrent_reads": number_of_concurrent_reads,
+                            "number_of_concurrent_writes": number_of_concurrent_writes,
+                            "number_of_failed_bulk_operations": number_of_failed_bulk_operations,
+                            "number_of_failed_fetches": number_of_failed_fetches,
+                            "number_of_operations_indexed": number_of_operations_indexed,
+                            "number_of_queued_writes": number_of_queued_writes,
+                            "number_of_successful_bulk_operations": number_of_successful_bulk_operations,
+                            "number_of_successful_fetches": number_of_successful_fetches,
+                            "operations_received": operations_received,
+                            "time_since_last_fetch_mills": time_since_last_fetch_millis,
+                            "total_fetch_time_millis": total_fetch_time_millis,
+                            "total_index_time_millis": total_index_time_millis,
+                            "total_transferred_bytes": total_transferred_bytes
+                        }
+                    ]
                 }
             ]
         }
@@ -333,13 +344,13 @@ class CcrStatsRecorderTests(TestCase):
 
         shard_metadata = {
             "cluster": "remote",
-            "index": "follower",
+            "index": follower_index,
             "shard": shard_id,
             "name": "ccr-stats"
         }
 
         metrics_store_put_doc.assert_called_with(
-            ccr_stats_follower_response["follower"][0],
+            ccr_stats_follower_response["indices"][0]["shards"][0],
             level=MetaInfoScope.cluster,
             meta_data=shard_metadata
         )
@@ -348,6 +359,8 @@ class CcrStatsRecorderTests(TestCase):
     def test_stores_default_ccr_stats_many_shards(self, metrics_store_put_doc):
         java_signed_maxlong = CcrStatsRecorderTests.java_signed_maxlong
 
+        leader_index="leader_cluster:leader"
+        follower_index="follower"
         shard_range = range(2)
         leader_global_checkpoint = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         leader_max_seq_no = [random.randint(0, java_signed_maxlong) for _ in shard_range]
@@ -357,40 +370,49 @@ class CcrStatsRecorderTests(TestCase):
         number_of_concurrent_reads = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_concurrent_writes = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_queued_writes = [random.randint(0, java_signed_maxlong) for _ in shard_range]
-        index_metadata_version = [random.randint(0, java_signed_maxlong) for _ in shard_range]
+        mapping_version = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         total_fetch_time_millis = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_successful_fetches = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_failed_fetches = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         operations_received = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         total_transferred_bytes = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         total_index_time_millis = [random.randint(0, java_signed_maxlong) for _ in shard_range]
+        time_since_last_fetch_millis = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_successful_bulk_operations = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_failed_bulk_operations = [random.randint(0, java_signed_maxlong) for _ in shard_range]
         number_of_operations_indexed = [random.randint(0, java_signed_maxlong) for _ in shard_range]
 
         ccr_stats_follower_response = {
-            "follower": [
+            "indices": [
                 {
-                    "shard_id": shard_id,
-                    "follower_global_checkpoint": follower_global_checkpoint[shard_id],
-                    "follower_max_seq_no": follower_max_seq_no[shard_id],
-                    "index_metadata_version": index_metadata_version[shard_id],
-                    "last_requested_seq_no": last_requested_seq_no[shard_id],
-                    "leader_global_checkpoint": leader_global_checkpoint[shard_id],
-                    "leader_max_seq_no": leader_max_seq_no[shard_id],
-                    "number_of_concurrent_reads": number_of_concurrent_reads[shard_id],
-                    "number_of_concurrent_writes": number_of_concurrent_writes[shard_id],
-                    "number_of_failed_bulk_operations": number_of_failed_bulk_operations[shard_id],
-                    "number_of_failed_fetches": number_of_failed_fetches[shard_id],
-                    "number_of_operations_indexed": number_of_operations_indexed[shard_id],
-                    "number_of_queued_writes": number_of_queued_writes[shard_id],
-                    "number_of_successful_bulk_operations": number_of_successful_bulk_operations[shard_id],
-                    "number_of_successful_fetches": number_of_successful_fetches[shard_id],
-                    "operations_received": operations_received[shard_id],
-                    "total_fetch_time_millis": total_fetch_time_millis[shard_id],
-                    "total_index_time_millis": total_index_time_millis[shard_id],
-                    "total_transferred_bytes": total_transferred_bytes[shard_id]
-                } for shard_id in shard_range
+                    "index": follower_index,
+                    "shards": [
+                        {
+                            "shard_id": shard_id,
+                            "follower_global_checkpoint": follower_global_checkpoint[shard_id],
+                            "leader_index": leader_index,
+                            "follower_index": follower_index,
+                            "follower_max_seq_no": follower_max_seq_no[shard_id],
+                            "mapping_version": mapping_version[shard_id],
+                            "last_requested_seq_no": last_requested_seq_no[shard_id],
+                            "leader_global_checkpoint": leader_global_checkpoint[shard_id],
+                            "leader_max_seq_no": leader_max_seq_no[shard_id],
+                            "number_of_concurrent_reads": number_of_concurrent_reads[shard_id],
+                            "number_of_concurrent_writes": number_of_concurrent_writes[shard_id],
+                            "number_of_failed_bulk_operations": number_of_failed_bulk_operations[shard_id],
+                            "number_of_failed_fetches": number_of_failed_fetches[shard_id],
+                            "number_of_operations_indexed": number_of_operations_indexed[shard_id],
+                            "number_of_queued_writes": number_of_queued_writes[shard_id],
+                            "number_of_successful_bulk_operations": number_of_successful_bulk_operations[shard_id],
+                            "number_of_successful_fetches": number_of_successful_fetches[shard_id],
+                            "operations_received": operations_received[shard_id],
+                            "total_fetch_time_millis": total_fetch_time_millis[shard_id],
+                            "total_index_time_millis": total_index_time_millis[shard_id],
+                            "total_transferred_bytes": total_transferred_bytes[shard_id],
+                            "time_since_last_fetch_millis": time_since_last_fetch_millis[shard_id]
+                        } for shard_id in shard_range
+                    ]
+                }
             ]
         }
 
@@ -416,8 +438,8 @@ class CcrStatsRecorderTests(TestCase):
         ]
 
         metrics_store_put_doc.assert_has_calls([
-            mock.call(ccr_stats_follower_response["follower"][0], level=MetaInfoScope.cluster, meta_data=shard_metadata[0]),
-            mock.call(ccr_stats_follower_response["follower"][1], level=MetaInfoScope.cluster, meta_data=shard_metadata[1])
+            mock.call(ccr_stats_follower_response["indices"][0]["shards"][0], level=MetaInfoScope.cluster, meta_data=shard_metadata[0]),
+            mock.call(ccr_stats_follower_response["indices"][0]["shards"][1], level=MetaInfoScope.cluster, meta_data=shard_metadata[1])
             ],
             any_order=True
         )
@@ -426,70 +448,89 @@ class CcrStatsRecorderTests(TestCase):
     def test_stores_filtered_ccr_stats(self, metrics_store_put_doc):
         java_signed_maxlong = CcrStatsRecorderTests.java_signed_maxlong
 
-        leader_global_checkpoint = random.randint(0, java_signed_maxlong),
-        leader_max_seq_no = random.randint(0, java_signed_maxlong),
-        follower_global_checkpoint = random.randint(0, java_signed_maxlong),
-        follower_max_seq_no = random.randint(0, java_signed_maxlong),
-        last_requested_seq_no = random.randint(0, java_signed_maxlong),
-        number_of_concurrent_reads = random.randint(0, java_signed_maxlong),
-        number_of_concurrent_writes = random.randint(0, java_signed_maxlong),
-        number_of_queued_writes = random.randint(0, java_signed_maxlong),
-        index_metadata_version = random.randint(0, java_signed_maxlong),
-        total_fetch_time_millis = random.randint(0, java_signed_maxlong),
-        number_of_successful_fetches = random.randint(0, java_signed_maxlong),
-        number_of_failed_fetches = random.randint(0, java_signed_maxlong),
-        operations_received = random.randint(0, java_signed_maxlong),
-        total_transferred_bytes = random.randint(0, java_signed_maxlong),
-        total_index_time_millis = random.randint(0, java_signed_maxlong),
-        number_of_successful_bulk_operations = random.randint(0, java_signed_maxlong),
-        number_of_failed_bulk_operations = random.randint(0, java_signed_maxlong),
+        leader_global_checkpoint = random.randint(0, java_signed_maxlong)
+        leader_max_seq_no = random.randint(0, java_signed_maxlong)
+        follower_global_checkpoint = random.randint(0, java_signed_maxlong)
+        leader_index1 = "leader_cluster:leader1"
+        follower_index1 = "follower1"
+        leader_index2 = "leader_cluster:leader2"
+        follower_index2 = "follower2"
+        follower_max_seq_no = random.randint(0, java_signed_maxlong)
+        last_requested_seq_no = random.randint(0, java_signed_maxlong)
+        number_of_concurrent_reads = random.randint(0, java_signed_maxlong)
+        number_of_concurrent_writes = random.randint(0, java_signed_maxlong)
+        number_of_queued_writes = random.randint(0, java_signed_maxlong)
+        mapping_version = random.randint(0, java_signed_maxlong)
+        total_fetch_time_millis = random.randint(0, java_signed_maxlong)
+        number_of_successful_fetches = random.randint(0, java_signed_maxlong)
+        number_of_failed_fetches = random.randint(0, java_signed_maxlong)
+        operations_received = random.randint(0, java_signed_maxlong)
+        total_transferred_bytes = random.randint(0, java_signed_maxlong)
+        total_index_time_millis = random.randint(0, java_signed_maxlong)
+        time_since_last_fetch_millis = random.randint(0, java_signed_maxlong)
+        number_of_successful_bulk_operations = random.randint(0, java_signed_maxlong)
+        number_of_failed_bulk_operations = random.randint(0, java_signed_maxlong)
         number_of_operations_indexed = random.randint(0, java_signed_maxlong)
 
         ccr_stats_follower_response = {
-            "follower1": [
+            "indices": [
                 {
-                    "shard_id": 0,
-                    "follower_global_checkpoint": follower_global_checkpoint,
-                    "follower_max_seq_no": follower_max_seq_no,
-                    "index_metadata_version": index_metadata_version,
-                    "last_requested_seq_no": last_requested_seq_no,
-                    "leader_global_checkpoint": leader_global_checkpoint,
-                    "leader_max_seq_no": leader_max_seq_no,
-                    "number_of_concurrent_reads": number_of_concurrent_reads,
-                    "number_of_concurrent_writes": number_of_concurrent_writes,
-                    "number_of_failed_bulk_operations": number_of_failed_bulk_operations,
-                    "number_of_failed_fetches": number_of_failed_fetches,
-                    "number_of_operations_indexed": number_of_operations_indexed,
-                    "number_of_queued_writes": number_of_queued_writes,
-                    "number_of_successful_bulk_operations": number_of_successful_bulk_operations,
-                    "number_of_successful_fetches": number_of_successful_fetches,
-                    "operations_received": operations_received,
-                    "total_fetch_time_millis": total_fetch_time_millis,
-                    "total_index_time_millis": total_index_time_millis,
-                    "total_transferred_bytes": total_transferred_bytes
-                }
-            ],
-            "follower2": [
+                    "index": "follower1",
+                    "shards": [
+                        {
+                            "shard_id": 0,
+                            "leader_index": leader_index1,
+                            "follower_index": follower_index1,
+                            "follower_global_checkpoint": follower_global_checkpoint,
+                            "follower_max_seq_no": follower_max_seq_no,
+                            "mapping_version": mapping_version,
+                            "last_requested_seq_no": last_requested_seq_no,
+                            "leader_global_checkpoint": leader_global_checkpoint,
+                            "leader_max_seq_no": leader_max_seq_no,
+                            "number_of_concurrent_reads": number_of_concurrent_reads,
+                            "number_of_concurrent_writes": number_of_concurrent_writes,
+                            "number_of_failed_bulk_operations": number_of_failed_bulk_operations,
+                            "number_of_failed_fetches": number_of_failed_fetches,
+                            "number_of_operations_indexed": number_of_operations_indexed,
+                            "number_of_queued_writes": number_of_queued_writes,
+                            "number_of_successful_bulk_operations": number_of_successful_bulk_operations,
+                            "number_of_successful_fetches": number_of_successful_fetches,
+                            "operations_received": operations_received,
+                            "time_since_last_fetch_millis": time_since_last_fetch_millis,
+                            "total_fetch_time_millis": total_fetch_time_millis,
+                            "total_index_time_millis": total_index_time_millis,
+                            "total_transferred_bytes": total_transferred_bytes
+                        }
+                    ]
+                },
                 {
-                    "shard_id": 1,
-                    "follower_global_checkpoint": follower_global_checkpoint,
-                    "follower_max_seq_no": follower_max_seq_no,
-                    "index_metadata_version": index_metadata_version,
-                    "last_requested_seq_no": last_requested_seq_no,
-                    "leader_global_checkpoint": leader_global_checkpoint,
-                    "leader_max_seq_no": leader_max_seq_no,
-                    "number_of_concurrent_reads": number_of_concurrent_reads,
-                    "number_of_concurrent_writes": number_of_concurrent_writes,
-                    "number_of_failed_bulk_operations": number_of_failed_bulk_operations,
-                    "number_of_failed_fetches": number_of_failed_fetches,
-                    "number_of_operations_indexed": number_of_operations_indexed,
-                    "number_of_queued_writes": number_of_queued_writes,
-                    "number_of_successful_bulk_operations": number_of_successful_bulk_operations,
-                    "number_of_successful_fetches": number_of_successful_fetches,
-                    "operations_received": operations_received,
-                    "total_fetch_time_millis": total_fetch_time_millis,
-                    "total_index_time_millis": total_index_time_millis,
-                    "total_transferred_bytes": total_transferred_bytes
+                    "index": "follower2",
+                    "shards": [
+                        {
+                            "shard_id": 1,
+                            "leader_index": leader_index2,
+                            "follower_index": follower_index2,
+                            "follower_global_checkpoint": follower_global_checkpoint,
+                            "follower_max_seq_no": follower_max_seq_no,
+                            "mapping_version": mapping_version,
+                            "last_requested_seq_no": last_requested_seq_no,
+                            "leader_global_checkpoint": leader_global_checkpoint,
+                            "leader_max_seq_no": leader_max_seq_no,
+                            "number_of_concurrent_reads": number_of_concurrent_reads,
+                            "number_of_concurrent_writes": number_of_concurrent_writes,
+                            "number_of_failed_bulk_operations": number_of_failed_bulk_operations,
+                            "number_of_failed_fetches": number_of_failed_fetches,
+                            "number_of_operations_indexed": number_of_operations_indexed,
+                            "number_of_queued_writes": number_of_queued_writes,
+                            "number_of_successful_bulk_operations": number_of_successful_bulk_operations,
+                            "number_of_successful_fetches": number_of_successful_fetches,
+                            "operations_received": operations_received,
+                            "time_since_last_fetch_millis": time_since_last_fetch_millis,
+                            "total_fetch_time_millis": total_fetch_time_millis,
+                            "total_index_time_millis": total_index_time_millis,
+                            "total_transferred_bytes": total_transferred_bytes
+                        }
+                        ]
                 }
             ]
         }
@@ -497,18 +538,18 @@ class CcrStatsRecorderTests(TestCase):
         client = Client(transport_client=TransportClient(response=ccr_stats_follower_response))
         cfg = create_config()
         metrics_store = metrics.EsMetricsStore(cfg)
-        recorder = telemetry.CcrStatsRecorder("remote", client, metrics_store, 1, indices=["follower1"])
+        recorder = telemetry.CcrStatsRecorder("remote", client, metrics_store, 1, indices=[follower_index1])
         recorder.record()
 
         shard_metadata = {
             "cluster": "remote",
-            "index": "follower1",
+            "index": follower_index1,
             "shard": 0,
             "name": "ccr-stats"
         }
 
         metrics_store_put_doc.assert_has_calls([
-            mock.call(ccr_stats_follower_response["follower1"][0], level=MetaInfoScope.cluster, meta_data=shard_metadata)
+            mock.call(ccr_stats_follower_response["indices"][0]["shards"][0], level=MetaInfoScope.cluster, meta_data=shard_metadata)
             ],
             any_order=True
         )
@@ -2046,19 +2087,21 @@ class MlBucketProcessingTimeTests(TestCase):
         metrics_store_put_doc.assert_has_calls([
             mock.call(doc={
                 "name": "ml_processing_time",
-                "job_name": "benchmark_ml_job_1",
-                "min_millis": 2.2,
-                "mean_millis": 12.3,
-                "median_millis": 17.2,
-                "max_millis": 36.0
+                "job": "benchmark_ml_job_1",
+                "min": 2.2,
+                "mean": 12.3,
+                "median": 17.2,
+                "max": 36.0,
+                "unit": "ms"
             }, level=metrics.MetaInfoScope.cluster),
             mock.call(doc={
                 "name": "ml_processing_time",
-                "job_name": "benchmark_ml_job_2",
-                "min_millis": 32.2,
-                "mean_millis": 78.3,
-                "median_millis": 37.4,
-                "max_millis": 226.3
+                "job": "benchmark_ml_job_2",
+                "min": 32.2,
+                "mean": 78.3,
+                "median": 37.4,
+                "max": 226.3,
+                "unit": "ms"
             }, level=metrics.MetaInfoScope.cluster)
         ])
 
diff --git a/tests/reporter_test.py b/tests/reporter_test.py
index fc22ece3..4739949e 100644
--- a/tests/reporter_test.py
+++ b/tests/reporter_test.py
@@ -47,11 +47,12 @@ class StatsCalculatorTests(TestCase):
                                       meta_data={"success": True})
         store.put_doc(doc={
             "name": "ml_processing_time",
-            "job_name": "benchmark_ml_job_1",
-            "min_millis": 2.2,
-            "mean_millis": 12.3,
-            "median_millis": 17.2,
-            "max_millis": 36.0
+            "job": "benchmark_ml_job_1",
+            "min": 2.2,
+            "mean": 12.3,
+            "median": 17.2,
+            "max": 36.0,
+            "unit": "ms"
         }, level=metrics.MetaInfoScope.cluster)
         store.put_count_node_level("rally-node-0", "final_index_size_bytes", 2048, unit="bytes")
         store.put_count_node_level("rally-node-1", "final_index_size_bytes", 4096, unit="bytes")
@@ -68,16 +69,17 @@ class StatsCalculatorTests(TestCase):
 
         self.assertEqual(6144, stats.index_size)
         self.assertEqual(1, len(stats.ml_processing_time))
-        self.assertEqual("benchmark_ml_job_1", stats.ml_processing_time[0]["job_name"])
-        self.assertEqual(2.2, stats.ml_processing_time[0]["min_millis"])
-        self.assertEqual(12.3, stats.ml_processing_time[0]["mean_millis"])
-        self.assertEqual(17.2, stats.ml_processing_time[0]["median_millis"])
-        self.assertEqual(36.0, stats.ml_processing_time[0]["max_millis"])
+        self.assertEqual("benchmark_ml_job_1", stats.ml_processing_time[0]["job"])
+        self.assertEqual(2.2, stats.ml_processing_time[0]["min"])
+        self.assertEqual(12.3, stats.ml_processing_time[0]["mean"])
+        self.assertEqual(17.2, stats.ml_processing_time[0]["median"])
+        self.assertEqual(36.0, stats.ml_processing_time[0]["max"])
+        self.assertEqual("ms", stats.ml_processing_time[0]["unit"])
 
 
-def select(l, name, operation=None, node=None):
+def select(l, name, operation=None, job=None, node=None):
     for item in l:
-        if item["name"] == name and item.get("operation") == operation and item.get("node") == node:
+        if item["name"] == name and item.get("operation") == operation and item.get("node") == node and item.get("job") == job:
             return item
     return None
 
@@ -116,6 +118,22 @@ class StatsTests(TestCase):
                     "startup_time": 4.2
                 }
             ],
+            "ml_processing_time": [
+                {
+                    "job": "job_1",
+                    "min": 3.3,
+                    "mean": 5.2,
+                    "median": 5.8,
+                    "max": 12.34
+                },
+                {
+                    "job": "job_2",
+                    "min": 3.55,
+                    "mean": 4.2,
+                    "median": 4.9,
+                    "max": 9.4
+                },
+            ],
             "young_gc_time": 68,
             "old_gc_time": 0,
             "refresh_time": 596,
@@ -188,6 +206,28 @@ class StatsTests(TestCase):
             }
         }, select(metric_list, "startup_time", node="rally-node-1"))
 
+        self.assertEqual({
+            "name": "ml_processing_time",
+            "job": "job_1",
+            "value": {
+                "min": 3.3,
+                "mean": 5.2,
+                "median": 5.8,
+                "max": 12.34
+            }
+        }, select(metric_list, "ml_processing_time", job="job_1"))
+
+        self.assertEqual({
+            "name": "ml_processing_time",
+            "job": "job_2",
+            "value": {
+                "min": 3.55,
+                "mean": 4.2,
+                "median": 4.9,
+                "max": 9.4
+            }
+        }, select(metric_list, "ml_processing_time", job="job_2"))
+
         self.assertEqual({
             "name": "young_gc_time",
             "value": {
diff --git a/tests/utils/repo_test.py b/tests/utils/repo_test.py
index ab67962c..73afa871 100644
--- a/tests/utils/repo_test.py
+++ b/tests/utils/repo_test.py
@@ -1,3 +1,4 @@
+import random
 from unittest import TestCase
 import unittest.mock as mock
 
@@ -112,7 +113,7 @@ class RallyRepositoryTests(TestCase):
             root_dir="/rally-resources",
             repo_name="unit-test",
             resource_name="unittest-resources",
-            offline=False)
+            offline=random.choice([True, False]))
 
         r.update(distribution_version="1.7.3")