[8.x] Add a new index setting to skip recovery source when synthetic …

…source is enabled (#118417) This change adds a new undocumented index settings that allows to use synthetic source for recovery and CCR without storing a recovery source.
elastic · Dec 11, 2024 · ceb53f7 · ceb53f7
1 parent 6946a3d
commit ceb53f7
Show file tree

Hide file tree

Showing 46 changed files with 1,993 additions and 895 deletions.
diff --git a/docs/changelog/114618.yaml b/docs/changelog/114618.yaml
@@ -0,0 +1,5 @@
+pr: 114618
+summary: Add a new index setting to skip recovery source when synthetic source is enabled
+area: Logs
+type: enhancement
+issues: []
diff --git a/.../internalClusterTest/java/org/elasticsearch/action/admin/indices/create/CloneIndexIT.java b/.../internalClusterTest/java/org/elasticsearch/action/admin/indices/create/CloneIndexIT.java
@@ -15,6 +15,7 @@
 import org.elasticsearch.common.ValidationException;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.query.TermsQueryBuilder;
 import org.elasticsearch.index.seqno.SeqNoStats;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -26,6 +27,7 @@
 import static org.elasticsearch.action.admin.indices.create.ShrinkIndexIT.assertNoResizeSourceIndexSettings;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
+import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 
@@ -143,6 +145,51 @@ public void testResizeChangeSyntheticSource() {
         assertThat(error.getMessage(), containsString("can't change setting [index.mapping.source.mode] during resize"));
     }
 
+    public void testResizeChangeRecoveryUseSyntheticSource() {
+        prepareCreate("source").setSettings(
+            indexSettings(between(1, 5), 0).put("index.mode", "logsdb")
+                .put(
+                    "index.version.created",
+                    IndexVersionUtils.randomVersionBetween(
+                        random(),
+                        IndexVersions.USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BACKPORT,
+                        IndexVersion.current()
+                    )
+                )
+        ).setMapping("@timestamp", "type=date", "host.name", "type=keyword").get();
+        updateIndexSettings(Settings.builder().put("index.blocks.write", true), "source");
+        IllegalArgumentException error = expectThrows(IllegalArgumentException.class, () -> {
+            indicesAdmin().prepareResizeIndex("source", "target")
+                .setResizeType(ResizeType.CLONE)
+                .setSettings(
+                    Settings.builder()
+                        .put(
+                            "index.version.created",
+                            IndexVersionUtils.randomVersionBetween(
+                                random(),
+                                IndexVersions.USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BACKPORT,
+                                IndexVersion.current()
+                            )
+                        )
+                        .put("index.recovery.use_synthetic_source", true)
+                        .put("index.mode", "logsdb")
+                        .putNull("index.blocks.write")
+                        .build()
+                )
+                .get();
+        });
+        // The index.recovery.use_synthetic_source setting requires either index.mode or index.mapping.source.mode
+        // to be present in the settings. Since these are all unmodifiable settings with a non-deterministic evaluation
+        // order, any of them may trigger a failure first.
+        assertThat(
+            error.getMessage(),
+            anyOf(
+                containsString("can't change setting [index.mode] during resize"),
+                containsString("can't change setting [index.recovery.use_synthetic_source] during resize")
+            )
+        );
+    }
+
     public void testResizeChangeIndexSorts() {
         prepareCreate("source").setSettings(indexSettings(between(1, 5), 0))
             .setMapping("@timestamp", "type=date", "host.name", "type=keyword")

diff --git a/server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java b/server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java
@@ -715,7 +715,15 @@ public void testShardChangesWithDefaultDocType() throws Exception {
         }
         IndexShard shard = indexService.getShard(0);
         try (
-            Translog.Snapshot luceneSnapshot = shard.newChangesSnapshot("test", 0, numOps - 1, true, randomBoolean(), randomBoolean());
+            Translog.Snapshot luceneSnapshot = shard.newChangesSnapshot(
+                "test",
+                0,
+                numOps - 1,
+                true,
+                randomBoolean(),
+                randomBoolean(),
+                randomLongBetween(1, ByteSizeValue.ofMb(32).getBytes())
+            );
             Translog.Snapshot translogSnapshot = getTranslog(shard).newSnapshot()
         ) {
             List<Translog.Operation> opsFromLucene = TestTranslog.drainSnapshot(luceneSnapshot, true);

diff --git a/server/src/internalClusterTest/java/org/elasticsearch/indices/recovery/IndexRecoveryIT.java b/server/src/internalClusterTest/java/org/elasticsearch/indices/recovery/IndexRecoveryIT.java
@@ -147,7 +147,6 @@
 import static org.elasticsearch.action.support.ActionTestUtils.assertNoFailureListener;
 import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
 import static org.elasticsearch.index.seqno.SequenceNumbers.NO_OPS_PERFORMED;
-import static org.elasticsearch.node.RecoverySettingsChunkSizePlugin.CHUNK_SIZE_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.empty;
@@ -247,7 +246,7 @@ private void assertOnGoingRecoveryState(
     public Settings.Builder createRecoverySettingsChunkPerSecond(long chunkSizeBytes) {
         return Settings.builder()
             // Set the chunk size in bytes
-            .put(CHUNK_SIZE_SETTING.getKey(), new ByteSizeValue(chunkSizeBytes, ByteSizeUnit.BYTES))
+            .put(RecoverySettings.INDICES_RECOVERY_CHUNK_SIZE.getKey(), new ByteSizeValue(chunkSizeBytes, ByteSizeUnit.BYTES))
             // Set one chunk of bytes per second.
             .put(RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING.getKey(), chunkSizeBytes, ByteSizeUnit.BYTES);
     }
@@ -270,7 +269,7 @@ private void unthrottleRecovery() {
             Settings.builder()
                 // 200mb is an arbitrary number intended to be large enough to avoid more throttling.
                 .put(RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING.getKey(), "200mb")
-                .put(CHUNK_SIZE_SETTING.getKey(), RecoverySettings.DEFAULT_CHUNK_SIZE)
+                .put(RecoverySettings.INDICES_RECOVERY_CHUNK_SIZE.getKey(), RecoverySettings.DEFAULT_CHUNK_SIZE)
         );
     }
 

diff --git a/server/src/internalClusterTest/java/org/elasticsearch/recovery/TruncatedRecoveryIT.java b/server/src/internalClusterTest/java/org/elasticsearch/recovery/TruncatedRecoveryIT.java
@@ -24,7 +24,7 @@
 import org.elasticsearch.indices.recovery.PeerRecoveryTargetService;
 import org.elasticsearch.indices.recovery.RecoveryFileChunkRequest;
 import org.elasticsearch.indices.recovery.RecoveryFilesInfoRequest;
-import org.elasticsearch.node.RecoverySettingsChunkSizePlugin;
+import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.transport.MockTransportService;
@@ -41,7 +41,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
-import static org.elasticsearch.node.RecoverySettingsChunkSizePlugin.CHUNK_SIZE_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -52,7 +51,7 @@ public class TruncatedRecoveryIT extends ESIntegTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Arrays.asList(MockTransportService.TestPlugin.class, RecoverySettingsChunkSizePlugin.class);
+        return Arrays.asList(MockTransportService.TestPlugin.class);
     }
 
     /**
@@ -63,7 +62,11 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
      */
     public void testCancelRecoveryAndResume() throws Exception {
         updateClusterSettings(
-            Settings.builder().put(CHUNK_SIZE_SETTING.getKey(), new ByteSizeValue(randomIntBetween(50, 300), ByteSizeUnit.BYTES))
+            Settings.builder()
+                .put(
+                    RecoverySettings.INDICES_RECOVERY_CHUNK_SIZE.getKey(),
+                    new ByteSizeValue(randomIntBetween(50, 300), ByteSizeUnit.BYTES)
+                )
         );
 
         NodesStatsResponse nodeStats = clusterAdmin().prepareNodesStats().get();

diff --git a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/RestoreSnapshotIT.java b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/RestoreSnapshotIT.java
@@ -800,6 +800,24 @@ public void testRestoreChangeSyntheticSource() {
         assertThat(error.getMessage(), containsString("cannot modify setting [index.mapping.source.mode] on restore"));
     }
 
+    public void testRestoreChangeRecoveryUseSyntheticSource() {
+        Client client = client();
+        createRepository("test-repo", "fs");
+        String indexName = "test-idx";
+        assertAcked(client.admin().indices().prepareCreate(indexName).setSettings(Settings.builder().put(indexSettings())));
+        createSnapshot("test-repo", "test-snap", Collections.singletonList(indexName));
+        cluster().wipeIndices(indexName);
+        var error = expectThrows(SnapshotRestoreException.class, () -> {
+            client.admin()
+                .cluster()
+                .prepareRestoreSnapshot(TEST_REQUEST_TIMEOUT, "test-repo", "test-snap")
+                .setIndexSettings(Settings.builder().put("index.recovery.use_synthetic_source", true))
+                .setWaitForCompletion(true)
+                .get();
+        });
+        assertThat(error.getMessage(), containsString("cannot modify setting [index.recovery.use_synthetic_source] on restore"));
+    }
+
     public void testRestoreChangeIndexSorts() {
         Client client = client();
         createRepository("test-repo", "fs");

diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java
@@ -1572,6 +1572,7 @@ static void validateCloneIndex(
     private static final Set<String> UNMODIFIABLE_SETTINGS_DURING_RESIZE = Set.of(
         IndexSettings.MODE.getKey(),
         SourceFieldMapper.INDEX_MAPPER_SOURCE_MODE_SETTING.getKey(),
+        IndexSettings.RECOVERY_USE_SYNTHETIC_SOURCE_SETTING.getKey(),
         IndexSortConfig.INDEX_SORT_FIELD_SETTING.getKey(),
         IndexSortConfig.INDEX_SORT_ORDER_SETTING.getKey(),
         IndexSortConfig.INDEX_SORT_MODE_SETTING.getKey(),

diff --git a/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java b/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java
@@ -257,6 +257,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         RecoverySettings.INDICES_RECOVERY_USE_SNAPSHOTS_SETTING,
         RecoverySettings.INDICES_RECOVERY_MAX_CONCURRENT_SNAPSHOT_FILE_DOWNLOADS,
         RecoverySettings.INDICES_RECOVERY_MAX_CONCURRENT_SNAPSHOT_FILE_DOWNLOADS_PER_NODE,
+        RecoverySettings.INDICES_RECOVERY_CHUNK_SIZE,
         RecoverySettings.NODE_BANDWIDTH_RECOVERY_FACTOR_READ_SETTING,
         RecoverySettings.NODE_BANDWIDTH_RECOVERY_FACTOR_WRITE_SETTING,
         RecoverySettings.NODE_BANDWIDTH_RECOVERY_OPERATOR_FACTOR_SETTING,

diff --git a/server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java b/server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java
@@ -189,6 +189,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         IgnoredSourceFieldMapper.SKIP_IGNORED_SOURCE_WRITE_SETTING,
         IgnoredSourceFieldMapper.SKIP_IGNORED_SOURCE_READ_SETTING,
         SourceFieldMapper.INDEX_MAPPER_SOURCE_MODE_SETTING,
+        IndexSettings.RECOVERY_USE_SYNTHETIC_SOURCE_SETTING,
 
         // validate that built-in similarities don't get redefined
         Setting.groupSetting("index.similarity.", (s) -> {

diff --git a/server/src/main/java/org/elasticsearch/index/IndexSettings.java b/server/src/main/java/org/elasticsearch/index/IndexSettings.java
@@ -38,6 +38,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -52,6 +53,7 @@
 import static org.elasticsearch.index.mapper.MapperService.INDEX_MAPPING_NESTED_DOCS_LIMIT_SETTING;
 import static org.elasticsearch.index.mapper.MapperService.INDEX_MAPPING_NESTED_FIELDS_LIMIT_SETTING;
 import static org.elasticsearch.index.mapper.MapperService.INDEX_MAPPING_TOTAL_FIELDS_LIMIT_SETTING;
+import static org.elasticsearch.index.mapper.SourceFieldMapper.INDEX_MAPPER_SOURCE_MODE_SETTING;
 
 /**
  * This class encapsulates all index level settings and handles settings updates.
@@ -656,6 +658,62 @@ public Iterator<Setting<?>> settings() {
         Property.Final
     );
 
+    public static final Setting<Boolean> RECOVERY_USE_SYNTHETIC_SOURCE_SETTING = Setting.boolSetting(
+        "index.recovery.use_synthetic_source",
+        false,
+        new Setting.Validator<>() {
+            @Override
+            public void validate(Boolean value) {}
+
+            @Override
+            public void validate(Boolean enabled, Map<Setting<?>, Object> settings) {
+                if (enabled == false) {
+                    return;
+                }
+
+                // Verify if synthetic source is enabled on the index; fail if it is not
+                var indexMode = (IndexMode) settings.get(MODE);
+                if (indexMode.defaultSourceMode() != SourceFieldMapper.Mode.SYNTHETIC) {
+                    var sourceMode = (SourceFieldMapper.Mode) settings.get(INDEX_MAPPER_SOURCE_MODE_SETTING);
+                    if (sourceMode != SourceFieldMapper.Mode.SYNTHETIC) {
+                        throw new IllegalArgumentException(
+                            String.format(
+                                Locale.ROOT,
+                                "The setting [%s] is only permitted when [%s] is set to [%s]. Current mode: [%s].",
+                                RECOVERY_USE_SYNTHETIC_SOURCE_SETTING.getKey(),
+                                INDEX_MAPPER_SOURCE_MODE_SETTING.getKey(),
+                                SourceFieldMapper.Mode.SYNTHETIC.name(),
+                                sourceMode.name()
+                            )
+                        );
+                    }
+                }
+
+                // Verify that all nodes can handle this setting
+                var version = (IndexVersion) settings.get(SETTING_INDEX_VERSION_CREATED);
+                if (version.before(IndexVersions.USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BACKPORT)) {
+                    throw new IllegalArgumentException(
+                        String.format(
+                            Locale.ROOT,
+                            "The setting [%s] is unavailable on this cluster because some nodes are running older "
+                                + "versions that do not support it. Please upgrade all nodes to the latest version "
+                                + "and try again.",
+                            RECOVERY_USE_SYNTHETIC_SOURCE_SETTING.getKey()
+                        )
+                    );
+                }
+            }
+
+            @Override
+            public Iterator<Setting<?>> settings() {
+                List<Setting<?>> res = List.of(INDEX_MAPPER_SOURCE_MODE_SETTING, SETTING_INDEX_VERSION_CREATED, MODE);
+                return res.iterator();
+            }
+        },
+        Property.IndexScope,
+        Property.Final
+    );
+
     /**
      * Returns <code>true</code> if TSDB encoding is enabled. The default is <code>true</code>
      */
@@ -827,6 +885,7 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
     private volatile boolean skipIgnoredSourceRead;
     private final SourceFieldMapper.Mode indexMappingSourceMode;
     private final boolean recoverySourceEnabled;
+    private final boolean recoverySourceSyntheticEnabled;
 
     /**
      * The maximum number of refresh listeners allows on this shard.
@@ -987,8 +1046,9 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         es87TSDBCodecEnabled = scopedSettings.get(TIME_SERIES_ES87TSDB_CODEC_ENABLED_SETTING);
         skipIgnoredSourceWrite = scopedSettings.get(IgnoredSourceFieldMapper.SKIP_IGNORED_SOURCE_WRITE_SETTING);
         skipIgnoredSourceRead = scopedSettings.get(IgnoredSourceFieldMapper.SKIP_IGNORED_SOURCE_READ_SETTING);
-        indexMappingSourceMode = scopedSettings.get(SourceFieldMapper.INDEX_MAPPER_SOURCE_MODE_SETTING);
+        indexMappingSourceMode = scopedSettings.get(INDEX_MAPPER_SOURCE_MODE_SETTING);
         recoverySourceEnabled = RecoverySettings.INDICES_RECOVERY_SOURCE_ENABLED_SETTING.get(nodeSettings);
+        recoverySourceSyntheticEnabled = scopedSettings.get(RECOVERY_USE_SYNTHETIC_SOURCE_SETTING);
 
         scopedSettings.addSettingsUpdateConsumer(
             MergePolicyConfig.INDEX_COMPOUND_FORMAT_SETTING,
@@ -1680,6 +1740,13 @@ public boolean isRecoverySourceEnabled() {
         return recoverySourceEnabled;
     }
 
+    /**
+     * @return Whether recovery source should always be bypassed in favor of using synthetic source.
+     */
+    public boolean isRecoverySourceSyntheticEnabled() {
+        return recoverySourceSyntheticEnabled;
+    }
+
     /**
      * The bounds for {@code @timestamp} on this index or
      * {@code null} if there are no bounds.

diff --git a/server/src/main/java/org/elasticsearch/index/IndexVersions.java b/server/src/main/java/org/elasticsearch/index/IndexVersions.java
@@ -122,6 +122,7 @@ private static IndexVersion def(int id, Version luceneVersion) {
     public static final IndexVersion LOGSDB_DEFAULT_IGNORE_DYNAMIC_BEYOND_LIMIT_BACKPORT = def(8_519_00_0, Version.LUCENE_9_12_0);
     public static final IndexVersion TIME_BASED_K_ORDERED_DOC_ID_BACKPORT = def(8_520_00_0, Version.LUCENE_9_12_0);
     public static final IndexVersion DEPRECATE_SOURCE_MODE_MAPPER = def(8_521_00_0, Version.LUCENE_9_12_0);
+    public static final IndexVersion USE_SYNTHETIC_SOURCE_FOR_RECOVERY_BACKPORT = def(8_522_00_0, Version.LUCENE_9_12_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

diff --git a/server/src/main/java/org/elasticsearch/index/engine/CombinedDocValues.java b/server/src/main/java/org/elasticsearch/index/engine/CombinedDocValues.java
@@ -24,6 +24,7 @@ final class CombinedDocValues {
     private final NumericDocValues primaryTermDV;
     private final NumericDocValues tombstoneDV;
     private final NumericDocValues recoverySource;
+    private final NumericDocValues recoverySourceSize;
 
     CombinedDocValues(LeafReader leafReader) throws IOException {
         this.versionDV = Objects.requireNonNull(leafReader.getNumericDocValues(VersionFieldMapper.NAME), "VersionDV is missing");
@@ -34,6 +35,7 @@ final class CombinedDocValues {
         );
         this.tombstoneDV = leafReader.getNumericDocValues(SeqNoFieldMapper.TOMBSTONE_NAME);
         this.recoverySource = leafReader.getNumericDocValues(SourceFieldMapper.RECOVERY_SOURCE_NAME);
+        this.recoverySourceSize = leafReader.getNumericDocValues(SourceFieldMapper.RECOVERY_SOURCE_SIZE_NAME);
     }
 
     long docVersion(int segmentDocId) throws IOException {
@@ -79,4 +81,12 @@ boolean hasRecoverySource(int segmentDocId) throws IOException {
         assert recoverySource.docID() < segmentDocId;
         return recoverySource.advanceExact(segmentDocId);
     }
+
+    long recoverySourceSize(int segmentDocId) throws IOException {
+        if (recoverySourceSize == null) {
+            return -1;
+        }
+        assert recoverySourceSize.docID() < segmentDocId;
+        return recoverySourceSize.advanceExact(segmentDocId) ? recoverySourceSize.longValue() : -1;
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/index/engine/Engine.java b/server/src/main/java/org/elasticsearch/index/engine/Engine.java
@@ -940,7 +940,7 @@ public boolean allowSearchIdleOptimization() {
      * @param source    the source of the request
      * @param fromSeqNo the start sequence number (inclusive)
      * @param toSeqNo   the end sequence number (inclusive)
-     * @see #newChangesSnapshot(String, long, long, boolean, boolean, boolean)
+     * @see #newChangesSnapshot(String, long, long, boolean, boolean, boolean, long)
      */
     public abstract int countChanges(String source, long fromSeqNo, long toSeqNo) throws IOException;
 
@@ -954,7 +954,8 @@ public abstract Translog.Snapshot newChangesSnapshot(
         long toSeqNo,
         boolean requiredFullRange,
         boolean singleConsumer,
-        boolean accessStats
+        boolean accessStats,
+        long maxChunkSize
     ) throws IOException;
 
     /**