TSDB ingest performance: combine routing and tsdb hashing

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,13 +62,16 @@
 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.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -321,15 +328,18 @@ public void testTsdbTemplatesNoKeywordFieldType() throws Exception {
                 .indexPatterns(List.of("k8s*"))
                 .template(
                     new Template(
-                        Settings.builder().put("index.mode", "time_series").put("index.routing_path", "metricset").build(),
+                        Settings.builder()
+                            .put("index.mode", "time_series")
+                            .put("index.routing_path", randomBoolean() ? "metricset" : null)
+                            .build(),
                         new CompressedXContent(mappingTemplate),
                         null
                     )
                 )
                 .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
                 .build()
         );
-        client().execute(TransportPutComposableIndexTemplateAction.TYPE, request).actionGet();
+        assertAcked(client().execute(TransportPutComposableIndexTemplateAction.TYPE, request));
     }
 
     public void testInvalidTsdbTemplatesMissingSettings() throws Exception {
@@ -619,6 +629,201 @@ public void testReindexing() throws Exception {
             getIndexResponse.getSetting(index2, IndexMetadata.INDEX_ROUTING_PATH.getKey()),
             equalTo(getIndexResponse.getSetting(index1, IndexMetadata.INDEX_ROUTING_PATH.getKey()))
         );
+        assertThat(
+            getIndexResponse.getSetting(index2, IndexMetadata.INDEX_DIMENSIONS.getKey()),
+            equalTo(getIndexResponse.getSetting(index1, IndexMetadata.INDEX_DIMENSIONS.getKey()))
+        );
+    }
+
+    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")));
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_ROUTING_PATH), 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"));
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_ROUTING_PATH), equalTo(List.of("metricset")));
+
+        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 custom metadata because the "labels" dynamic template doesn't have a path_math
+        assertThat(getSetting(dataStreamName, IndexMetadata.INDEX_DIMENSIONS), empty());
+        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_DIMENSIONS), empty());
+        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_DIMENSIONS), empty());
+        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_DIMENSIONS), empty());
+        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) {

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

@@ -185,6 +185,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

@@ -436,6 +436,10 @@ public void testSimulateTsdbDataStreamTemplate() throws Exception {
             ObjectPath.evaluate(responseBody, "template.settings.index.routing_path"),
             containsInAnyOrder("metricset", "k8s.pod.uid", "pod.labels.*")
         );
+        assertThat(
+            ObjectPath.evaluate(responseBody, "template.settings.index.dimensions"),
+            containsInAnyOrder("metricset", "k8s.pod.uid", "pod.labels.*")
+        );
         assertThat(ObjectPath.evaluate(responseBody, "overlapping"), empty());
     }