Make data streams a basic licensed feature. (#59392)

Backport of #59293 to 7.x branch.

* Create new data-stream xpack module.
* Move TimestampFieldMapper to the new module,
  this results in storing a composable index template
  with data stream definition only to work with default
  distribution. This way data streams can only be used
  with default distribution, since a data stream can
  currently only be created if a matching composable index
  template exists with a data stream definition.
* Renamed `_timestamp` meta field mapper
   to `_data_stream_timestamp` meta field mapper.
* Add logic to put composable index template api
  to fail if `_data_stream_timestamp` meta field mapper
  isn't registered. So that a more understandable
  error is returned when attempting to store a template
  with data stream definition via the oss distribution.

In a follow up the data stream transport and
rest actions can be moved to the xpack data-stream module.

Author: martijnvg

modules/rank-eval/src/test/resources/rest-api-spec/test/rank_eval/50_data_streams.yml

@@ -19,25 +19,19 @@
 
 package org.elasticsearch.upgrades;
 
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.StringEntity;
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.WarningFailureException;
-import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.MetadataIndexStateService;
-import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.CheckedFunction;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
@@ -56,7 +50,6 @@
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.Collection;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -1464,55 +1457,12 @@ public void testRecoveryWithTranslogRetentionDisabled() throws Exception {
         assertTotalHits(numDocs, entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search"))));
     }
 
-    @SuppressWarnings("unchecked")
-    public void testDataStreams() throws Exception {
-        assumeTrue("no data streams in versions before " + Version.V_7_9_0, getOldClusterVersion().onOrAfter(Version.V_7_9_0));
-        if (isRunningAgainstOldCluster()) {
-            String mapping = "{\n" +
-                "      \"properties\": {\n" +
-                "        \"@timestamp\": {\n" +
-                "          \"type\": \"date\"\n" +
-                "        }\n" +
-                "      }\n" +
-                "    }";
-            Template template = new Template(null, new CompressedXContent(mapping), null);
-            createComposableTemplate(client(), "dst", "ds", template);
-
-            Request indexRequest = new Request("POST", "/ds/_doc/1?op_type=create&refresh");
-            XContentBuilder builder = JsonXContent.contentBuilder().startObject()
-                .field("f", "v")
-                .field("@timestamp", new Date())
-                .endObject();
-            indexRequest.setJsonEntity(Strings.toString(builder));
-            assertOK(client().performRequest(indexRequest));
-        }
-
-        Request getDataStream = new Request("GET", "/_data_stream/ds");
-        Response response = client().performRequest(getDataStream);
-        assertOK(response);
-        List<Object> dataStreams = (List<Object>) entityAsMap(response).get("data_streams");
-        assertEquals(1, dataStreams.size());
-        Map<String, Object> ds = (Map<String, Object>) dataStreams.get(0);
-        List<Map<String, String>> indices = (List<Map<String, String>>) ds.get("indices");
-        assertEquals("ds", ds.get("name"));
-        assertEquals(1, indices.size());
-        assertEquals(DataStream.getDefaultBackingIndexName("ds", 1), indices.get(0).get("index_name"));
-        assertTotalHits(1, entityAsMap(client().performRequest(new Request("GET", "/ds/_search"))));
+    public static void assertNumHits(String index, int numHits, int totalShards) throws IOException {
+        Map<String, Object> resp = entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search")));
+        assertNoFailures(resp);
+        assertThat(XContentMapValues.extractValue("_shards.total", resp), equalTo(totalShards));
+        assertThat(XContentMapValues.extractValue("_shards.successful", resp), equalTo(totalShards));
+        assertThat(extractTotalHits(resp), equalTo(numHits));
     }
 
-    private static void createComposableTemplate(RestClient client, String templateName, String indexPattern, Template template)
-        throws IOException {
-        XContentBuilder builder = jsonBuilder();
-        template.toXContent(builder, ToXContent.EMPTY_PARAMS);
-        StringEntity templateJSON = new StringEntity(
-            String.format(Locale.ROOT, "{\n" +
-                "  \"index_patterns\": \"%s\",\n" +
-                "  \"data_stream\": { \"timestamp_field\": \"@timestamp\" },\n" +
-                "  \"template\": %s\n" +
-                "}", indexPattern, Strings.toString(builder)),
-            ContentType.APPLICATION_JSON);
-        Request createIndexTemplateRequest = new Request("PUT", "_index_template/" + templateName);
-        createIndexTemplateRequest.setEntity(templateJSON);
-        client.performRequest(createIndexTemplateRequest);
-    }
 }

qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java

@@ -3,47 +3,6 @@
   - skip:
       features: allowed_warnings
 
-  - do:
-      allowed_warnings:
-        - "index template [my-template1] has index patterns [simple-data-stream1] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
-      indices.put_index_template:
-        name: my-template1
-        body:
-          index_patterns: [simple-data-stream1]
-          template:
-            mappings:
-              properties:
-                '@timestamp':
-                  type: date
-          data_stream:
-            timestamp_field: '@timestamp'
-  - do:
-      allowed_warnings:
-        - "index template [my-template2] has index patterns [simple-data-stream2] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template2] will take precedence during new index creation"
-      indices.put_index_template:
-        name: my-template2
-        body:
-          index_patterns: [simple-data-stream2]
-          template:
-            mappings:
-              properties:
-                '@timestamp':
-                  type: date
-          data_stream:
-            timestamp_field: '@timestamp'
-
-  - do:
-      indices.create_data_stream:
-        name: simple-data-stream1
-
-  - do:
-      indices.create_data_stream:
-        name: simple-data-stream2
-
-  - do:
-      indices.rollover:
-        alias: "simple-data-stream2"
-
   - do:
       indices.create:
         index: single_doc_index
@@ -123,18 +82,6 @@
                       nested2:
                         type: keyword
                         doc_values: false
-
-  - do:
-      indices.create:
-        index: closed_index
-        body:
-          aliases:
-            aliased_closed_index: {}
-
-  - do:
-      indices.close:
-        index: closed_index
-
   - do:
       indices.create:
         index: test_index

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/remote_cluster/10_basic.yml

@@ -109,47 +109,3 @@ setup:
           settings:
             index.number_of_replicas: 0
             index.number_of_shards: 6
-
----
-"Prohibit clone on data stream's write index":
-  - skip:
-      version: " - 7.8.99"
-      reason: "data streams only supported in 7.9+"
-      features: allowed_warnings
-
-  - do:
-      allowed_warnings:
-        - "index template [my-template1] has index patterns [simple-data-stream1] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
-      indices.put_index_template:
-        name: my-template1
-        body:
-          index_patterns: [simple-data-stream1]
-          template:
-            mappings:
-              properties:
-                '@timestamp':
-                  type: date
-          data_stream:
-            timestamp_field: '@timestamp'
-
-  - do:
-      indices.create_data_stream:
-        name: simple-data-stream1
-  - is_true: acknowledged
-
-  - do:
-      catch: bad_request
-      indices.clone:
-        index: ".ds-simple-data-stream1-000001"
-        target: "target"
-        wait_for_active_shards: 1
-        master_timeout: 10s
-        body:
-          settings:
-            index.number_of_replicas: 0
-            index.number_of_shards: 2
-
-  - do:
-      indices.delete_data_stream:
-        name: simple-data-stream1
-  - is_true: acknowledged