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

docs/configuration.rst

@@ -80,6 +80,8 @@ The following settings are applicable only if ``datastore.type`` is set to "elas
 * ``datastore.probe.cluster_version`` (default: true): Enables automatic detection of the metric store's version.
 * ``datastore.number_of_shards`` (default: `Elasticsearch default value <https://www.elastic.co/guide/en/elasticsearch/reference/current/index-modules.html#_static_index_settings>`_): The number of primary shards that the ``rally-*`` indices should have. Any updates to this setting after initial index creation will only be applied to new ``rally-*`` indices.
 * ``datastore.number_of_replicas`` (default: `Elasticsearch default value <https://www.elastic.co/guide/en/elasticsearch/reference/current/index-modules.html#_static_index_settings>`_): The number of replicas each primary shard has. Defaults to . Any updates to this setting after initial index creation will only be applied to new ``rally-*`` indices.
+* ``datastore.overwrite_existing_templates`` (default: ``false``): Existing Rally index templates are replaced only when this option is ``true``.
+
 
 **Examples**
 

docs/migrate.rst

@@ -9,6 +9,12 @@ Minimum Python version is 3.9.0
 
 Rally 2.12.0 requires Python 3.9.0 or above. Check the :ref:`updated installation instructions <install_python>` for more details.
 
+The metrics store keeps existing index templates
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Existing Rally index templates are replaced only when option ``datastore.overwrite_existing_templates`` in section ``reporting`` is ``true``.
+
+
 Migrating to Rally 2.10.1
 -------------------------
 

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,7 @@ 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())
+            self._ensure_index_template()
             if not self._client.exists(index=self._index):
                 self._client.create_index(index=self._index)
             else:
@@ -913,6 +916,34 @@ def open(self, race_id=None, race_timestamp=None, track_name=None, challenge_nam
         # ensure we can search immediately after opening
         self._client.refresh(index=self._index)
 
+    def _ensure_index_template(self):
+        new_template: str = self._get_template()
+
+        old_template: dict | None = None
+        if self._client.template_exists("rally-metrics"):
+            for t in self._client.get_template("rally-metrics").body.get("index_templates", []):
+                old_template = t.get("index_template", {}).get("template", {})
+                break
+
+        if old_template is None:
+            self.logger.info(
+                "Create index template:\n%s",
+                pretty.dump(json.loads(new_template).get("template", {}), pretty.Flag.FLAT_DICT),
+            )
+        else:
+            diff = pretty.diff(old_template, json.loads(new_template).get("template", {}), pretty.Flag.FLAT_DICT)
+            if diff == "":
+                self.logger.debug("Keep existing template (it is identical)")
+                return
+            if not convert.to_bool(
+                self._config.opts(section="reporting", key="datastore.overwrite_existing_templates", default_value=False, mandatory=False)
+            ):
+                self.logger.debug("Keep existing template (datastore.overwrite_existing_templates = false):\n%s", diff)
+                return
+            self.logger.warning("Overwrite existing index template (datastore.overwrite_existing_templates = true):\n%s", diff)
+
+        self._client.put_template("rally-metrics", new_template)
+
     def index_name(self):
         ts = time.from_iso8601(self._race_timestamp)
         return "rally-metrics-%04d-%02d" % (ts.year, ts.month)

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

pyproject.toml

@@ -120,7 +120,7 @@ develop = [
     "black==24.10.0",
     # mypy
     "boto3-stubs==1.26.125",
-    "mypy==1.10.1",
+    "mypy==1.15.0",
     "types-psutil==5.9.4",
     "types-tabulate==0.8.9",
     "types-urllib3==1.26.19",

tests/metrics_test.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 # pylint: disable=protected-access
+from __future__ import annotations
 
 import datetime
 import json
@@ -27,17 +28,55 @@
 from dataclasses import dataclass
 from unittest import mock
 
+import elastic_transport
 import elasticsearch.exceptions
 import elasticsearch.helpers
 import pytest
 
 from esrally import config, exceptions, metrics, paths, track
 from esrally.metrics import GlobalStatsCalculator
 from esrally.track import Challenge, Operation, Task, Track
-from esrally.utils import opts
+from esrally.utils import cases, opts, pretty
+
+
+def rally_metric_template():
+    return {
+        "mappings": {
+            "_source": {"enabled": True},
+            "date_detection": False,
+            "dynamic_templates": [
+                {"strings": {"mapping": {"ignore_above": 8191, "type": "keyword"}, "match": "*", "match_mapping_type": "string"}}
+            ],
+            "properties": {
+                "@timestamp": {"format": "epoch_millis", "type": "date"},
+                "car": {"type": "keyword"},
+                "challenge": {"type": "keyword"},
+                "environment": {"type": "keyword"},
+                "job": {"type": "keyword"},
+                "max": {"type": "float"},
+                "mean": {"type": "float"},
+                "median": {"type": "float"},
+                "meta": {"properties": {"error-description": {"type": "wildcard"}}},
+                "min": {"type": "float"},
+                "name": {"type": "keyword"},
+                "operation": {"type": "keyword"},
+                "operation-type": {"type": "keyword"},
+                "race-id": {"type": "keyword"},
+                "race-timestamp": {"fields": {"raw": {"type": "keyword"}}, "format": "basic_date_time_no_millis", "type": "date"},
+                "relative-time": {"type": "float"},
+                "sample-type": {"type": "keyword"},
+                "task": {"type": "keyword"},
+                "track": {"type": "keyword"},
+                "unit": {"type": "keyword"},
+                "value": {"type": "float"},
+            },
+        },
+        "settings": {"index": {"mapping": {"total_fields": {"limit": "2000"}}, "number_of_replicas": "3", "number_of_shards": "3"}},
+    }
 
 
 class MockClientFactory:
+
     def __init__(self, cfg):
         self._es = mock.create_autospec(metrics.EsClient)
 
@@ -49,8 +88,8 @@ class DummyIndexTemplateProvider:
     def __init__(self, cfg):
         pass
 
-    def metrics_template(self):
-        return "metrics-test-template"
+    def metrics_template(self) -> str:
+        return json.dumps({"index_patterns": ["rally-metrics-*"], "template": provided_metrics_template()})
 
     def races_template(self):
         return "races-test-template"
@@ -59,6 +98,12 @@ def results_template(self):
         return "results-test-template"
 
 
+def provided_metrics_template() -> dict:
+    template = rally_metric_template()
+    template["settings"]["index"] = {"mapping.total_fields.limit": 2000, "number_of_shards": 1, "number_of_replicas": 1}
+    return template
+
+
 class StaticClock:
     NOW = 1453362707
 
@@ -438,6 +483,67 @@ def setup_method(self, method):
         # get hold of the mocked client...
         self.es_mock = self.metrics_store._client
         self.es_mock.exists.return_value = False
+        self.es_mock.template_exists.return_value = False
+        self.es_mock.get_template.return_value = mock.create_autospec(elastic_transport.ObjectApiResponse, body={"index_templates": []})
+        self.metrics_store.logger = mock.create_autospec(logging.Logger)
+
+    @dataclass
+    class OpenCase:
+        create: bool = True
+        template: dict | None = None
+        overwrite_templates: str | None = None
+        want_put_template: bool = False
+        want_logger_call: mock._Call | None = None
+
+    @cases.cases(
+        create_false=OpenCase(create=False),
+        default=OpenCase(
+            want_put_template=True,
+            want_logger_call=mock.call.info("Create index template:\n%s", pretty.dump(provided_metrics_template(), pretty.Flag.FLAT_DICT)),
+        ),
+        template_exists=OpenCase(
+            template=rally_metric_template(),
+            want_logger_call=mock.call.debug(
+                "Keep existing template (datastore.overwrite_existing_templates = false):\n%s",
+                pretty.diff(rally_metric_template(), provided_metrics_template(), pretty.Flag.FLAT_DICT),
+            ),
+        ),
+        keep_identical_template=OpenCase(
+            template=provided_metrics_template(), want_logger_call=mock.call.debug("Keep existing template (it is identical)")
+        ),
+        overwrite_templates_true=OpenCase(
+            template=rally_metric_template(),
+            overwrite_templates="true",
+            want_put_template=True,
+            want_logger_call=mock.call.warning(
+                "Overwrite existing index template (datastore.overwrite_existing_templates = true):\n%s",
+                pretty.diff(rally_metric_template(), provided_metrics_template(), pretty.Flag.FLAT_DICT),
+            ),
+        ),
+        overwrite_templates_false=OpenCase(
+            template=rally_metric_template(),
+            overwrite_templates="false",
+            want_logger_call=mock.call.debug(
+                "Keep existing template (datastore.overwrite_existing_templates = false):\n%s",
+                pretty.diff(rally_metric_template(), provided_metrics_template(), pretty.Flag.FLAT_DICT),
+            ),
+        ),
+    )
+    def test_open(self, case: OpenCase):
+        if case.template is not None:
+            self.metrics_store._client.template_exists.return_value = True
+            self.metrics_store._client.get_template.return_value.body["index_templates"] = [{"index_template": {"template": case.template}}]
+        if case.overwrite_templates is not None:
+            self.cfg.add(
+                scope=config.Scope.application,
+                section="reporting",
+                key="datastore.overwrite_existing_templates",
+                value=case.overwrite_templates,
+            )
+        self.metrics_store.open(self.RACE_ID, self.RACE_TIMESTAMP, "test", "append", "defaults", create=case.create)
+        assert case.want_put_template == self.metrics_store._client.put_template.called
+        if case.want_logger_call is not None:
+            assert self.metrics_store.logger.method_calls[-1:] == [case.want_logger_call]
 
     def test_put_value_without_meta_info(self):
         throughput = 5000

tests/types_test.py

@@ -126,5 +126,5 @@ def assert_annotations(obj, ident, *expects):
 class TestConfigTypeHint:
     def test_esrally_module_annotations(self):
         for module in project_root.glob_modules("esrally/**/*.py"):
-            assert_annotations(module, "cfg", types.Config)
+            assert_annotations(module, "cfg", types.Config, "types.Config")
             assert_annotations(module, "config", types.Config, Optional[types.Config], ConfigParser)