Preserve pre-existing index templates (#1900).

When opening an `EsMetricsStore`, it creates the index template if any of following:
 - the index template doesn't exist
 - `reporting/datastore.overwrite_existing_templates` option is `true`

It will preserve existing template on all the other cases.
It adds a new method for getting boolean configuration options.
It logs a warning when an existing index template is being replaced.
It highlights index template differences between the existing one (if any) and the
configured one (according to rally.ini).

esrally/config.py

@@ -14,6 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
 
 import configparser
 import logging
@@ -25,6 +26,9 @@
 from esrally import PROGRAM_NAME, exceptions, paths, types
 from esrally.utils import io
 
+# Copied from configparser for back compatibility
+BOOLEAN_STATES: dict[str, bool] = {"1": True, "yes": True, "true": True, "on": True, "0": False, "no": False, "false": False, "off": False}
+
 
 class Scope(Enum):
     # Valid for all benchmarks, typically read from the configuration file
@@ -185,6 +189,25 @@ def opts(self, section: types.Section, key: types.Key, default_value=None, manda
             else:
                 raise exceptions.ConfigError(f"No value for mandatory configuration: section='{section}', key='{key}'")
 
+    def boolean(self, section: types.Section, key: types.Key, default: bool | None = None) -> bool:
+        """It reads a boolean value from the configuration.
+
+        :param section: section represents the section name in the configuration.
+        :param key: section represents the option name in the configuration.
+        :param default: default specifies a value to be returned when any is found in the configuration. If it is None
+            and any is found in the configuration then ConfigError is raised.
+        :return: the boolean value if found in the configuration or the default value otherwise if not None.
+        """
+        default_value: str | None = None
+        if default is not None:
+            default_value = str(default).lower()
+        value = self.opts(section=section, key=key, default_value=default_value, mandatory=default_value is None)
+        if isinstance(value, str):
+            value = BOOLEAN_STATES.get(value)
+        if value is None:
+            raise exceptions.ConfigError(f"Invalid value for boolean option: section='{section}', key='{key}', value='{value}'")
+        return bool(value)
+
     def all_opts(self, section: types.Section):
         """
         Finds all options in a section and returns them in a dict.

esrally/metrics.py

@@ -14,6 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+# from __future__ import annotations
 
 import collections
 import datetime
@@ -33,7 +34,7 @@
 import tabulate
 
 from esrally import client, config, exceptions, paths, time, types, version
-from esrally.utils import console, convert, io, versions
+from esrally.utils import console, convert, io, pretty, versions
 
 
 class EsClient:
@@ -47,6 +48,9 @@ def __init__(self, client, cluster_version=None):
         self._cluster_version = cluster_version
         self.retryable_status_codes = [502, 503, 504, 429]
 
+    def get_template(self, name):
+        return self.guarded(self._client.indices.get_index_template, name=name)
+
     def put_template(self, name, template):
         tmpl = json.loads(template)
         return self.guarded(self._client.indices.put_index_template, name=name, **tmpl)
@@ -898,8 +902,31 @@ def open(self, race_id=None, race_timestamp=None, track_name=None, challenge_nam
         self._index = self.index_name()
         # reduce a bit of noise in the metrics cluster log
         if create:
-            # always update the mapping to the latest version
-            self._client.put_template("rally-metrics", self._get_template())
+            template = None
+            if self._client.template_exists("rally-metrics"):
+                for t in self._client.get_template("rally-metrics").body.get("index_templates", []):
+                    template = t.get("index_template", {}).get("template", {})
+                    break
+
+            new_template: str = self._get_template()
+            if template is None or self._config.boolean(section="reporting", key="datastore.overwrite_existing_templates", default=False):
+                if template is None:
+                    self.logger.info(
+                        "Create index template:\n%s",
+                        pretty.diff(old={}, new=json.loads(new_template).get("template", {}), flat_dict=True),
+                    )
+                else:
+                    self.logger.warning(
+                        "Overwrite existing index template (datastore.overwrite_existing_templates = true):\n%s",
+                        pretty.diff(old=template, new=json.loads(new_template).get("template", {}), flat_dict=True),
+                    )
+                self._client.put_template("rally-metrics", new_template)
+            else:
+                self.logger.warning(
+                    "Keep existing template (datastore.overwrite_existing_templates = false):\n%s",
+                    pretty.diff(old=template, new=json.loads(new_template).get("template", {}), flat_dict=True),
+                )
+
             if not self._client.exists(index=self._index):
                 self._client.create_index(index=self._index)
             else:

esrally/types.py

@@ -66,6 +66,7 @@
     "datastore.host",
     "datastore.number_of_replicas",
     "datastore.number_of_shards",
+    "datastore.overwrite_existing_templates",
     "datastore.password",
     "datastore.port",
     "datastore.probe.cluster_version",

esrally/utils/cases.py

@@ -0,0 +1,51 @@
+# Licensed to Elasticsearch B.V. under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Elasticsearch B.V. licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from typing import Callable, Protocol
+
+import pytest
+
+
+class Case(Protocol):
+    """Case represent the interface expected an instance passed to cases decorator is expected to implement."""
+
+    # case_id specifies the ID of the test case.
+    case_id: str
+
+
+def cases(*cases: Case) -> Callable:
+    """cases defines a decorator wrapping pytest.mark.parametrize to run a test with multiple cases.
+
+    Example of use:
+
+        @dataclass
+        class SumCase:
+            id: str
+            values: list[int]
+            want: int
+
+        @cases(
+            MyCase("no_values", want=0)
+            MyCase("2 values", values=[1,2] want=3),
+        )
+        def test_sum(case: MyCase):
+            assert sum(*case.values) == case.want
+
+    :param cases: sequence of cases
+    :return: test method decorator
+    """
+    return pytest.mark.parametrize(argnames="case", argvalues=cases, ids=lambda case: case.case_id)

esrally/utils/pretty.py

@@ -0,0 +1,70 @@
+# Licensed to Elasticsearch B.V. under one or more contributor
+# license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright
+# ownership. Elasticsearch B.V. licenses this file to you under
+# the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import difflib
+import json
+import typing
+from collections import abc
+
+O = typing.Union[abc.Mapping, abc.Sequence, str, int, float, bool, None]
+
+
+_diff_methods = []
+
+
+def diff(old, new: O, flat_dict=False) -> str:
+    return "\n".join(difflib.ndiff(_dump(old, flat_dict=flat_dict), _dump(new, flat_dict=flat_dict)))
+
+
+def dump(o: O, flat_dict=False):
+    return "\n".join(_dump(o, flat_dict=flat_dict))
+
+
+def _dump(o: O, flat_dict=False) -> abc.Sequence[str]:
+    if flat_dict:
+        o = _flat_dict(o)
+    return json.dumps(o, indent=2, sort_keys=True).splitlines()
+
+
+def _flat_dict(o: O) -> dict[str, str]:
+    """Given a JSON like object, it produces a key value flat dictionary of strings easy to read and compare.
+    :param o: a JSON like object
+    :return: a flat dictionary
+    """
+    return dict(_visit_nested(o))
+
+
+def _visit_nested(o: O) -> abc.Generator[tuple[str, str], None, None]:
+    if isinstance(o, (str, bytes)):
+        yield "", str(o)
+    elif isinstance(o, abc.Mapping):
+        for k1, v1 in o.items():
+            for k2, v2 in _visit_nested(v1):
+                if k2:
+                    yield f"{k1}.{k2}", v2
+                else:
+                    yield k1, v2
+    elif isinstance(o, abc.Sequence):
+        for k1, v1 in enumerate(o):
+            for k2, v2 in _visit_nested(v1):
+                if k2:
+                    yield f"{k1}.{k2}", v2
+                else:
+                    yield str(k1), v2
+    else:
+        yield "", json.dumps(o)

tests/config_test.py

@@ -15,11 +15,15 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from __future__ import annotations
+
 import configparser
+from dataclasses import dataclass
 
 import pytest
 
 from esrally import config, exceptions
+from esrally.utils.cases import cases
 
 
 class MockInput:
@@ -241,6 +245,37 @@ def test_can_migrate_outdated_config(self):
     def assert_equals_base_config(self, base_config, local_config, section, key):
         assert base_config.opts(section, key) == local_config.opts(section, key)
 
+    @dataclass
+    class BooleanCase:
+        case_id: str
+        value: str | None = None
+        default: bool | None = None
+        want: bool | None = None
+        want_error: type[Exception] = None
+
+    @cases(
+        BooleanCase(
+            "no-value",
+            want_error=exceptions.ConfigError,
+        ),
+        BooleanCase("value-false", value="false", want=False),
+        BooleanCase("value-true", value="true", want=True),
+        BooleanCase("default-false", default=False, want=False),
+        BooleanCase("default-true", default=True, want=True),
+    )
+    def test_boolean(self, case):
+        cfg = config.Config()
+        if case.value is not None:
+            cfg.add(scope=None, section="reporting", key="values", value=case.value)
+        try:
+            got = cfg.boolean(section="reporting", key="values", default=case.default)
+        except Exception as ex:
+            assert isinstance(ex, case.want_error)
+            assert case.want is None
+        else:
+            assert case.want == got
+            assert case.want_error is None
+
 
 class TestConfigMigration:
     def test_does_not_migrate_outdated_config(self):