TSDB ingest performance: combine routing and tsdb hashing

gradle/verification-metadata.xml

@@ -271,6 +271,11 @@
             <sha256 value="a3609eeb7173837a589a4ad865e6feaf71ee6139d061eae2e698401485f7589c" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="com.dynatrace.hash4j" name="hash4j" version="0.25.0">
+         <artifact name="hash4j-0.25.0.jar">
+            <sha256 value="a9f1ebf6dab3fecfb36eed4e084ca587e5cdf9dcdd723b8adb7d17b6fc874862" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="com.ethlo.time" name="itu" version="1.7.0">
          <artifact name="itu-1.7.0.jar">
             <sha256 value="55ceb418c9e8138c4fcf62e213c4c814d89e8a84c827d395407cbecba5d791e7" origin="Generated by Gradle"/>

modules/data-streams/build.gradle

@@ -52,6 +52,10 @@ tasks.named("yamlRestCompatTestTransform").configure({ task ->
   task.skipTest("data_stream/30_auto_create_data_stream/Don't initialize failure store during data stream auto-creation on successful index", "Configuring the failure store via data stream templates is not supported anymore.")
 
   task.skipTest("data_stream/150_tsdb/TSDB failures go to failure store", "Configuring the failure store via data stream templates is not supported anymore.")
+  // TODO remove these after removing exact _tsid assertions in 8.x
+  task.skipTest("data_stream/150_tsdb/dynamic templates", "The _tsid has changed in a new index version. This tests verifies the exact _tsid value with is too brittle for compatibility testing.")
+  task.skipTest("data_stream/150_tsdb/dynamic templates - conflicting aliases", "The _tsid has changed in a new index version. This tests verifies the exact _tsid value with is too brittle for compatibility testing.")
+  task.skipTest("data_stream/150_tsdb/dynamic templates with nesting", "The _tsid has changed in a new index version. This tests verifies the exact _tsid value with is too brittle for compatibility testing.")
 
   task.skipTest("data_stream/170_modify_data_stream/Modify a data stream's failure store", "Configuring the failure store via data stream templates is not supported anymore.")
 

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/TSDBIndexingIT.java

@@ -14,6 +14,8 @@
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
 import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
+import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
+import org.elasticsearch.action.admin.indices.mapping.put.TransportPutMappingAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
 import org.elasticsearch.action.admin.indices.segments.IndicesSegmentsRequest;
@@ -24,6 +26,7 @@
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.bulk.IndexDocFailureStoreStatus;
+import org.elasticsearch.action.datastreams.CreateDataStreamAction;
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
@@ -34,6 +37,7 @@
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.compress.CompressedXContent;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.time.FormatNames;
@@ -58,12 +62,14 @@
 import java.time.temporal.ChronoUnit;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 
 import static org.elasticsearch.test.MapMatcher.assertMap;
 import static org.elasticsearch.test.MapMatcher.matchesMap;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertResponse;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -320,16 +326,12 @@ public void testTsdbTemplatesNoKeywordFieldType() throws Exception {
             ComposableIndexTemplate.builder()
                 .indexPatterns(List.of("k8s*"))
                 .template(
-                    new Template(
-                        Settings.builder().put("index.mode", "time_series").put("index.routing_path", "metricset").build(),
-                        new CompressedXContent(mappingTemplate),
-                        null
-                    )
+                    new Template(Settings.builder().put("index.mode", "time_series").build(), new CompressedXContent(mappingTemplate), null)
                 )
                 .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
                 .build()
         );
-        client().execute(TransportPutComposableIndexTemplateAction.TYPE, request).actionGet();
+        assertResponse(client().execute(TransportPutComposableIndexTemplateAction.TYPE, request), ElasticsearchAssertions::assertAcked);
     }
 
     public void testInvalidTsdbTemplatesMissingSettings() throws Exception {
@@ -621,6 +623,192 @@ public void testReindexing() throws Exception {
         );
     }
 
+    public void testAddDimensionToMapping() throws Exception {
+        String dataStreamName = "my-ds";
+        var putTemplateRequest = new TransportPutComposableIndexTemplateAction.Request("id");
+        putTemplateRequest.indexTemplate(
+            ComposableIndexTemplate.builder()
+                .indexPatterns(List.of(dataStreamName))
+                .template(
+                    new Template(
+                        Settings.builder().put("index.mode", "time_series").build(),
+                        new CompressedXContent(MAPPING_TEMPLATE),
+                        null
+                    )
+                )
+                .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
+                .build()
+        );
+        assertAcked(client().execute(TransportPutComposableIndexTemplateAction.TYPE, putTemplateRequest));
+
+        // create data stream
+        CreateDataStreamAction.Request createDsRequest = new CreateDataStreamAction.Request(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            "my-ds"
+        );
+        assertAcked(client().execute(CreateDataStreamAction.INSTANCE, createDsRequest));
+
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_DIMENSIONS), equalTo(List.of("metricset")));
+
+        // put mapping with k8s.pod.uid as another time series dimension
+        var putMappingRequest = new PutMappingRequest(dataStreamName).source("""
+            {
+              "properties": {
+                "k8s.pod.name": {
+                  "type": "keyword",
+                  "time_series_dimension": true
+                }
+              }
+            }
+            """, XContentType.JSON);
+        assertAcked(client().execute(TransportPutMappingAction.TYPE, putMappingRequest).actionGet());
+
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_DIMENSIONS), containsInAnyOrder("metricset", "k8s.pod.name"));
+
+        indexWithPodNames(dataStreamName, Instant.now(), Map.of(), "dog", "cat");
+    }
+
+    public void testDynamicStringDimensions() throws Exception {
+        String dataStreamName = "my-ds";
+        var putTemplateRequest = new TransportPutComposableIndexTemplateAction.Request("id");
+        putTemplateRequest.indexTemplate(
+            ComposableIndexTemplate.builder()
+                .indexPatterns(List.of(dataStreamName))
+                .template(new Template(Settings.builder().put("index.mode", "time_series").build(), new CompressedXContent("""
+
+                        {
+                      "_doc": {
+                        "dynamic_templates": [
+                          {
+                            "labels": {
+                              "match_mapping_type": "string",
+                              "mapping": {
+                                "type": "keyword",
+                                "time_series_dimension": true
+                              }
+                            }
+                          }
+                        ],
+                        "properties": {
+                          "@timestamp": {
+                            "type": "date"
+                          },
+                          "metricset": {
+                            "type": "keyword",
+                            "time_series_dimension": true
+                          }
+                        }
+                      }
+                    }"""), null))
+                .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
+                .build()
+        );
+        assertAcked(client().execute(TransportPutComposableIndexTemplateAction.TYPE, putTemplateRequest));
+
+        CreateDataStreamAction.Request createDsRequest = new CreateDataStreamAction.Request(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            "my-ds"
+        );
+        assertAcked(client().execute(CreateDataStreamAction.INSTANCE, createDsRequest));
+
+        // doesn't populate index.dimensions because the "labels" dynamic template doesn't have a path_math
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_ROUTING_PATH), equalTo(List.of("metricset")));
+
+        // index doc
+        BulkResponse bulkResponse = client().prepareBulk()
+            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+            .add(
+                client().prepareIndex(dataStreamName)
+                    .setOpType(DocWriteRequest.OpType.CREATE)
+                    .setSource(DOC.replace("$time", formatInstant(Instant.now())), XContentType.JSON)
+            )
+            .get();
+        assertThat(bulkResponse.hasFailures(), is(false));
+
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_ROUTING_PATH), equalTo(List.of("metricset")));
+    }
+
+    public void testDynamicDimensions() throws Exception {
+        String dataStreamName = "my-ds";
+        var putTemplateRequest = new TransportPutComposableIndexTemplateAction.Request("id");
+        putTemplateRequest.indexTemplate(
+            ComposableIndexTemplate.builder()
+                .indexPatterns(List.of(dataStreamName))
+                .template(new Template(Settings.builder().put("index.mode", "time_series").build(), new CompressedXContent("""
+
+                        {
+                      "_doc": {
+                        "dynamic_templates": [
+                          {
+                            "label": {
+                              "mapping": {
+                                "type": "keyword",
+                                "time_series_dimension": true
+                              }
+                            }
+                          }
+                        ],
+                        "properties": {
+                          "@timestamp": {
+                            "type": "date"
+                          },
+                          "metricset": {
+                            "type": "keyword",
+                            "time_series_dimension": true
+                          }
+                        }
+                      }
+                    }"""), null))
+                .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
+                .build()
+        );
+        assertAcked(client().execute(TransportPutComposableIndexTemplateAction.TYPE, putTemplateRequest));
+
+        CreateDataStreamAction.Request createDsRequest = new CreateDataStreamAction.Request(
+            TEST_REQUEST_TIMEOUT,
+            TEST_REQUEST_TIMEOUT,
+            "my-ds"
+        );
+        assertAcked(client().execute(CreateDataStreamAction.INSTANCE, createDsRequest));
+
+        // doesn't populate index.dimensions because the "label" dynamic template doesn't have a path_math
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_ROUTING_PATH), equalTo(List.of("metricset")));
+
+        // index doc
+        indexWithPodNames(dataStreamName, Instant.now(), Map.of("k8s.pod.name", "label"), "dog", "cat");
+
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_ROUTING_PATH), equalTo(List.of("metricset")));
+    }
+
+    private void indexWithPodNames(String dataStreamName, Instant timestamp, Map<String, String> dynamicTemplates, String... podNames) {
+        // index doc
+        BulkRequestBuilder bulkRequestBuilder = client().prepareBulk();
+        bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+        for (String podName : podNames) {
+            bulkRequestBuilder.add(
+                client().prepareIndex(dataStreamName)
+                    .setOpType(DocWriteRequest.OpType.CREATE)
+                    .setSource(DOC.replace("$time", formatInstant(timestamp)).replace("dog", podName), XContentType.JSON)
+                    .request()
+                    .setDynamicTemplates(dynamicTemplates)
+            );
+        }
+
+        BulkResponse bulkResponse = bulkRequestBuilder.get();
+        assertThat(bulkResponse.hasFailures(), is(false));
+    }
+
+    private <T> T getSetting(String dataStreamName, Setting<T> setting) {
+        GetIndexResponse getIndexResponse = safeGet(
+            indicesAdmin().getIndex(new GetIndexRequest(TEST_REQUEST_TIMEOUT).indices(dataStreamName))
+        );
+        assertThat(getIndexResponse.getIndices().length, equalTo(1));
+        Settings settings = getIndexResponse.getSettings().get(getIndexResponse.getIndices()[0]);
+        return setting.get(settings);
+    }
+
     static String formatInstant(Instant instant) {
         return DateFormatter.forPattern(FormatNames.STRICT_DATE_OPTIONAL_TIME.getName()).format(instant);
     }

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/TSDBPassthroughIndexingIT.java

@@ -184,7 +184,7 @@ public void testIndexingGettingAndSearching() throws Exception {
 
         // validate index:
         var getIndexResponse = client().admin().indices().getIndex(new GetIndexRequest(TEST_REQUEST_TIMEOUT).indices(index)).actionGet();
-        assertThat(getIndexResponse.getSettings().get(index).get("index.routing_path"), equalTo("[attributes.*]"));
+        assertThat(getIndexResponse.getSettings().get(index).get("index.dimensions"), equalTo("[attributes.*]"));
         // validate mapping
         var mapping = getIndexResponse.mappings().get(index).getSourceAsMap();
         assertMap(

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/TsdbDataStreamRestIT.java

@@ -433,9 +433,10 @@ public void testSimulateTsdbDataStreamTemplate() throws Exception {
         assertThat(ObjectPath.evaluate(responseBody, "template.settings.index.time_series.start_time"), notNullValue());
         assertThat(ObjectPath.evaluate(responseBody, "template.settings.index.time_series.end_time"), notNullValue());
         assertThat(
-            ObjectPath.evaluate(responseBody, "template.settings.index.routing_path"),
+            ObjectPath.evaluate(responseBody, "template.settings.index.dimensions"),
             containsInAnyOrder("metricset", "k8s.pod.uid", "pod.labels.*")
         );
+        assertThat(ObjectPath.evaluate(responseBody, "template.settings.index.routing_path"), nullValue());
         assertThat(ObjectPath.evaluate(responseBody, "overlapping"), empty());
     }