Introduce a setting controlling the activation of the logs index mode in logs@settings

docs/changelog/109025.yaml

@@ -0,0 +1,7 @@
+pr: 109025
+summary: Introduce a feature flag and a setting controlling the activation of the
+  `logs` index mode
+area: Logs
+type: feature
+issues:
+ - 108762

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

@@ -62,7 +62,7 @@ public class LogsDataStreamIT extends ESSingleNodeTestCase {
               "@timestamp" : {
                 "type": "date"
               },
-              "hostname": {
+              "host.name": {
                 "type": "keyword"
               },
               "pid": {
@@ -86,7 +86,7 @@ public class LogsDataStreamIT extends ESSingleNodeTestCase {
               "@timestamp" : {
                 "type": "date"
               },
-              "hostname": {
+              "host.name": {
                 "type": "keyword",
                 "time_series_dimension": "true"
               },
@@ -110,7 +110,7 @@ public class LogsDataStreamIT extends ESSingleNodeTestCase {
     private static final String LOG_DOC_TEMPLATE = """
         {
             "@timestamp": "%s",
-            "hostname": "%s",
+            "host.name": "%s",
             "pid": "%d",
             "method": "%s",
             "message": "%s",
@@ -121,7 +121,7 @@ public class LogsDataStreamIT extends ESSingleNodeTestCase {
     private static final String TIME_SERIES_DOC_TEMPLATE = """
         {
             "@timestamp": "%s",
-            "hostname": "%s",
+            "host.name": "%s",
             "pid": "%d",
             "method": "%s",
             "ip_address": "%s",
@@ -207,7 +207,7 @@ public void testIndexModeLogsAndTimeSeriesSwitching() throws IOException, Execut
         final String dataStreamName = generateDataStreamName("custom");
         final List<String> indexPatterns = List.of("custom-*-*");
         final Map<String, String> logsSettings = Map.of("index.mode", "logs");
-        final Map<String, String> timeSeriesSettings = Map.of("index.mode", "time_series", "index.routing_path", "hostname");
+        final Map<String, String> timeSeriesSettings = Map.of("index.mode", "time_series", "index.routing_path", "host.name");
 
         putComposableIndexTemplate(client(), "custom-composable-template", LOGS_OR_STANDARD_MAPPING, logsSettings, indexPatterns);
         createDataStream(client(), dataStreamName);
@@ -224,7 +224,7 @@ public void testIndexModeLogsAndTimeSeriesSwitching() throws IOException, Execut
         assertDataStreamBackingIndicesModes(dataStreamName, List.of(IndexMode.LOGS, IndexMode.TIME_SERIES, IndexMode.LOGS));
     }
 
-    public void testInvalidIndexModeTimeSeriesSwitchWithoutROutingPath() throws IOException, ExecutionException, InterruptedException {
+    public void testInvalidIndexModeTimeSeriesSwitchWithoutRoutingPath() throws IOException, ExecutionException, InterruptedException {
         final String dataStreamName = generateDataStreamName("custom");
         final List<String> indexPatterns = List.of("custom-*-*");
         final Map<String, String> logsSettings = Map.of("index.mode", "logs");
@@ -250,7 +250,7 @@ public void testInvalidIndexModeTimeSeriesSwitchWithoutDimensions() throws IOExc
         final String dataStreamName = generateDataStreamName("custom");
         final List<String> indexPatterns = List.of("custom-*-*");
         final Map<String, String> logsSettings = Map.of("index.mode", "logs");
-        final Map<String, String> timeSeriesSettings = Map.of("index.mode", "time_series", "index.routing_path", "hostname");
+        final Map<String, String> timeSeriesSettings = Map.of("index.mode", "time_series", "index.routing_path", "host.name");
 
         putComposableIndexTemplate(client(), "custom-composable-template", LOGS_OR_STANDARD_MAPPING, logsSettings, indexPatterns);
         createDataStream(client(), dataStreamName);
@@ -269,8 +269,9 @@ public void testInvalidIndexModeTimeSeriesSwitchWithoutDimensions() throws IOExc
         assertThat(
             exception.getCause().getCause().getMessage(),
             Matchers.equalTo(
-                "All fields that match routing_path must be configured with [time_series_dimension: true] or flattened fields with "
-                    + "a list of dimensions in [time_series_dimensions] and without the [script] parameter. [hostname] was not a dimension."
+                "All fields that match routing_path must be configured with [time_series_dimension: true] or flattened fields "
+                    + "with a list of dimensions in [time_series_dimensions] and without the [script] parameter. [host.name] was not a "
+                    + "dimension."
             )
         );
     }

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

@@ -80,7 +80,7 @@ private static void waitForLogs(RestClient client) throws Exception {
                 "@timestamp" : {
                   "type": "date"
                 },
-                "hostname": {
+                "host.name": {
                   "type": "keyword"
                 },
                 "pid": {
@@ -116,7 +116,7 @@ private static void waitForLogs(RestClient client) throws Exception {
                   "@timestamp" : {
                     "type": "date"
                 },
-                "hostname": {
+                "host.name": {
                   "type": "keyword",
                   "time_series_dimension": "true"
                 },
@@ -138,7 +138,7 @@ private static void waitForLogs(RestClient client) throws Exception {
     private static final String DOC_TEMPLATE = """
         {
             "@timestamp": "%s",
-            "hostname": "%s",
+            "host.name": "%s",
             "pid": "%d",
             "method": "%s",
             "message": "%s",

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

@@ -0,0 +1,56 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams.logsdb;
+
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.hamcrest.Matchers;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+import java.io.IOException;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class LogsIndexModeDisabledRestTestIT extends LogsIndexModeRestTestIT {
+
+    @ClassRule()
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .distribution(DistributionType.DEFAULT)
+        .module("constant-keyword")
+        .module("data-streams")
+        .module("mapper-extras")
+        .module("x-pack-aggregate-metric")
+        .module("x-pack-stack")
+        .setting("xpack.security.enabled", "false")
+        .setting("xpack.license.self_generated.type", "trial")
+        .build();
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
+
+    @Before
+    public void setup() throws Exception {
+        client = client();
+        waitForLogs(client);
+    }
+
+    private RestClient client;
+
+    public void testLogsSettingsIndexModeDisabled() throws IOException {
+        assertOK(createDataStream(client, "logs-custom-dev"));
+        final String indexMode = (String) getSetting(client, getDataStreamBackingIndex(client, "logs-custom-dev", 0), "index.mode");
+        assertThat(indexMode, Matchers.not(equalTo(IndexMode.LOGS.getName())));
+    }
+
+}

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams.logsdb;
+
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.hamcrest.Matchers;
+import org.junit.Before;
+import org.junit.ClassRule;
+
+import java.io.IOException;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class LogsIndexModeEnabledRestTestIT extends LogsIndexModeRestTestIT {
+
+    @ClassRule()
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .distribution(DistributionType.DEFAULT)
+        .module("constant-keyword")
+        .module("data-streams")
+        .module("mapper-extras")
+        .module("x-pack-aggregate-metric")
+        .module("x-pack-stack")
+        .setting("xpack.security.enabled", "false")
+        .setting("xpack.license.self_generated.type", "trial")
+        .setting("cluster.logsdb.enabled", "true")
+        .build();
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
+
+    @Before
+    public void setup() throws Exception {
+        client = client();
+        waitForLogs(client);
+    }
+
+    private RestClient client;
+
+    private static final String MAPPINGS = """
+        {
+          "template": {
+            "mappings": {
+              "properties": {
+                "method": {
+                  "type": "keyword"
+                },
+                "message": {
+                  "type": "text"
+                }
+              }
+            }
+          }
+        }""";
+
+    private static String BULK_INDEX_REQUEST = """
+        { "create": {}}
+        { "@timestamp": "2023-01-01T05:11:00Z", "host.name": "foo", "method" : "PUT", "message": "foo put message" }
+        { "create": {}}
+        { "@timestamp": "2023-01-01T05:12:00Z", "host.name": "bar", "method" : "POST", "message": "bar post message" }
+        { "create": {}}
+        { "@timestamp": "2023-01-01T05:12:00Z", "host.name": "baz", "method" : "PUT", "message": "baz put message" }
+        { "create": {}}
+        { "@timestamp": "2023-01-01T05:13:00Z", "host.name": "baz", "method" : "PUT", "message": "baz put message" }
+        """;
+
+    public void testCreateDataStream() throws IOException {
+        assertOK(putComponentTemplate(client, "logs@custom", MAPPINGS));
+        assertOK(createDataStream(client, "logs-custom-dev"));
+        final String indexMode = (String) getSetting(client, getDataStreamBackingIndex(client, "logs-custom-dev", 0), "index.mode");
+        assertThat(indexMode, equalTo(IndexMode.LOGS.getName()));
+    }
+
+    public void testBulkIndexing() throws IOException {
+        assertOK(putComponentTemplate(client, "logs@custom", MAPPINGS));
+        assertOK(createDataStream(client, "logs-custom-dev"));
+        assertOK(bulkIndex(client, "logs-custom-dev", () -> BULK_INDEX_REQUEST));
+    }
+
+    public void testRolloverDataStream() throws IOException {
+        assertOK(putComponentTemplate(client, "logs@custom", MAPPINGS));
+        assertOK(createDataStream(client, "logs-custom-dev"));
+        final String firstBackingIndex = getDataStreamBackingIndex(client, "logs-custom-dev", 0);
+        assertOK(rolloverDataStream(client, "logs-custom-dev"));
+        final String secondBackingIndex = getDataStreamBackingIndex(client, "logs-custom-dev", 1);
+        assertThat(firstBackingIndex, Matchers.not(equalTo(secondBackingIndex)));
+        assertThat(getDataStreamBackingIndices(client, "logs-custom-dev").size(), equalTo(2));
+    }
+}

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams.logsdb;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.test.rest.ESRestTestCase;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+public abstract class LogsIndexModeRestTestIT extends ESRestTestCase {
+    protected static void waitForLogs(RestClient client) throws Exception {
+        assertBusy(() -> {
+            try {
+                final Request request = new Request("GET", "_index_template/logs");
+                assertOK(client.performRequest(request));
+            } catch (ResponseException e) {
+                fail(e.getMessage());
+            }
+        });
+    }
+
+    protected static Response putComponentTemplate(final RestClient client, final String templateName, final String mappings)
+        throws IOException {
+        final Request request = new Request("PUT", "/_component_template/" + templateName);
+        request.setJsonEntity(mappings);
+        return client.performRequest(request);
+    }
+
+    protected static Response createDataStream(final RestClient client, final String dataStreamName) throws IOException {
+        return client.performRequest(new Request("PUT", "_data_stream/" + dataStreamName));
+    }
+
+    protected static Response rolloverDataStream(final RestClient client, final String dataStreamName) throws IOException {
+        return client.performRequest(new Request("POST", "/" + dataStreamName + "/_rollover"));
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static String getDataStreamBackingIndex(final RestClient client, final String dataStreamName, int backingIndex)
+        throws IOException {
+        final Request request = new Request("GET", "_data_stream/" + dataStreamName);
+        final List<Object> dataStreams = (List<Object>) entityAsMap(client.performRequest(request)).get("data_streams");
+        final Map<String, Object> dataStream = (Map<String, Object>) dataStreams.get(0);
+        final List<Map<String, String>> backingIndices = (List<Map<String, String>>) dataStream.get("indices");
+        return backingIndices.get(backingIndex).get("index_name");
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static List<String> getDataStreamBackingIndices(final RestClient client, final String dataStreamName) throws IOException {
+        final Request request = new Request("GET", "_data_stream/" + dataStreamName);
+        final List<Object> dataStreams = (List<Object>) entityAsMap(client.performRequest(request)).get("data_streams");
+        final Map<String, Object> dataStream = (Map<String, Object>) dataStreams.get(0);
+        final List<Map<String, String>> backingIndices = (List<Map<String, String>>) dataStream.get("indices");
+        return backingIndices.stream().map(map -> map.get("indices")).collect(Collectors.toList());
+    }
+
+    @SuppressWarnings("unchecked")
+    protected static Object getSetting(final RestClient client, final String indexName, final String setting) throws IOException {
+        final Request request = new Request("GET", "/" + indexName + "/_settings?flat_settings=true&include_defaults=true");
+        final Map<String, Object> settings = ((Map<String, Map<String, Object>>) entityAsMap(client.performRequest(request)).get(indexName))
+            .get("settings");
+
+        return settings.get(setting);
+    }
+
+    protected static Response bulkIndex(final RestClient client, final String dataStreamName, final Supplier<String> bulkSupplier)
+        throws IOException {
+        var bulkRequest = new Request("POST", "/" + dataStreamName + "/_bulk");
+        bulkRequest.setJsonEntity(bulkSupplier.get());
+        bulkRequest.addParameter("refresh", "true");
+        return client().performRequest(bulkRequest);
+    }
+}