Enable ignore_malformed in logsdb (#113072) (#113077)

This change enables ignore_malformed by default for newly created 
logsdb indices.

Closes #106822

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

@@ -12,6 +12,7 @@ package org.elasticsearch.datastreams.logsdb;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.junit.Before;
@@ -284,6 +285,55 @@ public class LogsIndexModeCustomSettingsIT extends LogsIndexModeRestTestIT {
         assertThat(ignoreDynamicBeyondLimitIndexSetting, equalTo("false"));
     }
 
+    public void testIgnoreMalformedSetting() throws IOException {
+        // with default template
+        {
+            assertOK(createDataStream(client, "logs-test-1"));
+            String logsIndex1 = getDataStreamBackingIndex(client, "logs-test-1", 0);
+            assertThat(getSetting(client, logsIndex1, "index.mapping.ignore_malformed"), equalTo("true"));
+            for (String newValue : List.of("false", "true")) {
+                closeIndex(logsIndex1);
+                updateIndexSettings(logsIndex1, Settings.builder().put("index.mapping.ignore_malformed", newValue));
+                assertThat(getSetting(client, logsIndex1, "index.mapping.ignore_malformed"), equalTo(newValue));
+            }
+        }
+        // with override template
+        {
+            var template = """
+                {
+                  "template": {
+                    "settings": {
+                      "index": {
+                        "mapping": {
+                          "ignore_malformed": "false"
+                        }
+                      }
+                    }
+                  }
+                }""";
+            assertOK(putComponentTemplate(client, "logs@custom", template));
+            assertOK(createDataStream(client, "logs-custom-dev"));
+            String index = getDataStreamBackingIndex(client, "logs-custom-dev", 0);
+            assertThat(getSetting(client, index, "index.mapping.ignore_malformed"), equalTo("false"));
+            for (String newValue : List.of("true", "false")) {
+                closeIndex(index);
+                updateIndexSettings(index, Settings.builder().put("index.mapping.ignore_malformed", newValue));
+                assertThat(getSetting(client, index, "index.mapping.ignore_malformed"), equalTo(newValue));
+            }
+        }
+        // standard index
+        {
+            String index = "test-index";
+            createIndex(index);
+            assertThat(getSetting(client, index, "index.mapping.ignore_malformed"), equalTo("false"));
+            for (String newValue : List.of("false", "true")) {
+                closeIndex(index);
+                updateIndexSettings(index, Settings.builder().put("index.mapping.ignore_malformed", newValue));
+                assertThat(getSetting(client, index, "index.mapping.ignore_malformed"), equalTo(newValue));
+            }
+        }
+    }
+
     private static Map<String, Object> getMapping(final RestClient client, final String indexName) throws IOException {
         final Request request = new Request("GET", "/" + indexName + "/_mapping");
 

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

@@ -70,10 +70,14 @@ public abstract class LogsIndexModeRestTestIT extends ESRestTestCase {
     @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);
+        Map<String, Object> response = entityAsMap(client.performRequest(request));
+        final Map<String, Object> settings = ((Map<String, Map<String, Object>>) response.get(indexName)).get("settings");
+        final Map<String, Object> defaults = ((Map<String, Map<String, Object>>) response.get(indexName)).get("defaults");
+        Object val = settings.get(setting);
+        if (val == null) {
+            val = defaults.get(setting);
+        }
+        return val;
     }
 
     protected static Response bulkIndex(final RestClient client, final String dataStreamName, final Supplier<String> bulkSupplier)

server/src/main/java/org/elasticsearch/index/IndexVersions.java

@@ -114,6 +114,8 @@ public class IndexVersions {
     public static final IndexVersion UPGRADE_TO_LUCENE_9_11_1 = def(8_511_00_0, Version.LUCENE_9_11_1);
     public static final IndexVersion INDEX_SORTING_ON_NESTED = def(8_512_00_0, Version.LUCENE_9_11_1);
     public static final IndexVersion LENIENT_UPDATEABLE_SYNONYMS = def(8_513_00_0, Version.LUCENE_9_11_1);
+    public static final IndexVersion ENABLE_IGNORE_MALFORMED_LOGSDB = def(8_514_00_0, Version.LUCENE_9_11_1);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/index/mapper/FieldMapper.java

@@ -12,6 +12,7 @@ package org.elasticsearch.index.mapper;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.index.LeafReaderContext;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.Explicit;
 import org.elasticsearch.common.TriFunction;
 import org.elasticsearch.common.collect.Iterators;
@@ -22,6 +23,8 @@ import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.analysis.NamedAnalyzer;
@@ -60,12 +63,15 @@ import static org.elasticsearch.core.Strings.format;
 public abstract class FieldMapper extends Mapper {
     private static final Logger logger = LogManager.getLogger(FieldMapper.class);
 
-    public static final Setting<Boolean> IGNORE_MALFORMED_SETTING = Setting.boolSetting(
-        "index.mapping.ignore_malformed",
-        false,
-        Property.IndexScope,
-        Property.ServerlessPublic
-    );
+    public static final Setting<Boolean> IGNORE_MALFORMED_SETTING = Setting.boolSetting("index.mapping.ignore_malformed", settings -> {
+        if (IndexSettings.MODE.get(settings) == IndexMode.LOGSDB
+            && IndexMetadata.SETTING_INDEX_VERSION_CREATED.get(settings).onOrAfter(IndexVersions.ENABLE_IGNORE_MALFORMED_LOGSDB)) {
+            return "true";
+        } else {
+            return "false";
+        }
+    }, Property.IndexScope, Property.ServerlessPublic);
+
     public static final Setting<Boolean> COERCE_SETTING = Setting.boolSetting(
         "index.mapping.coerce",
         false,