Associate translog with Lucene index commit

Author: tlrx

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -166,6 +166,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
             IndexSettings.FINAL_PIPELINE,
             MetaDataIndexStateService.VERIFIED_BEFORE_CLOSE_SETTING,
             IndexSettings.ON_HEAP_ID_TERMS_INDEX,
+            IndexSettings.SKIP_FILES_RECOVERY_SETTING,
             ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING,
 
             // validate that built-in similarities don't get redefined

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

@@ -325,6 +325,13 @@ public final class IndexSettings {
     public static final Setting<Double> FILE_BASED_RECOVERY_THRESHOLD_SETTING
         = Setting.doubleSetting("index.recovery.file_based_threshold", 0.1d, 0.0d, Setting.Property.IndexScope);
 
+    /**
+     * A badly named setting indicating that for some specific shards we skip the files recovery and assume that the
+     * files are available.
+     */
+    public static final Setting<Boolean> SKIP_FILES_RECOVERY_SETTING =
+        Setting.boolSetting("index.recovery.skip_files", false, Setting.Property.IndexScope, Property.PrivateIndex);
+
     private final Index index;
     private final Version version;
     private final Logger logger;

server/src/main/java/org/elasticsearch/index/shard/StoreRecovery.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.EngineException;
 import org.elasticsearch.index.mapper.MapperService;
@@ -398,7 +399,11 @@ private void internalRecoverFromStore(IndexShard indexShard) throws IndexShardRe
                 writeEmptyRetentionLeasesFile(indexShard);
             } else if (indexShouldExists) {
                 if (recoveryState.getRecoverySource().shouldBootstrapNewHistoryUUID()) {
-                    store.bootstrapNewHistory();
+                    if (IndexSettings.SKIP_FILES_RECOVERY_SETTING.get(indexShard.indexSettings().getSettings())) {
+                        associateTranslogWithIndex(indexShard, store);
+                    } else {
+                        store.bootstrapNewHistory();
+                    }
                     writeEmptyRetentionLeasesFile(indexShard);
                 }
                 // since we recover from local, just fill the files and size
@@ -460,7 +465,14 @@ private void restore(IndexShard indexShard, Repository repository, SnapshotRecov
         final ActionListener<Void> restoreListener = ActionListener.wrap(
             v -> {
                 final Store store = indexShard.store();
-                bootstrap(indexShard, store);
+                if (IndexSettings.SKIP_FILES_RECOVERY_SETTING.get(indexShard.indexSettings().getSettings())) {
+                    // shard is restored from a snapshot and we expect the store to already contains the files,
+                    // hence we can skip bootstraping a new history uuid with a new translog, and simply
+                    // associate an empty translog with the existing lucene commit.
+                    associateTranslogWithIndex(indexShard, store);
+                } else {
+                    bootstrap(indexShard, store);
+                }
                 assert indexShard.shardRouting.primary() : "only primary shards can recover from store";
                 writeEmptyRetentionLeasesFile(indexShard);
                 indexShard.openEngineAndRecoverFromTranslog();
@@ -491,9 +503,18 @@ private void restore(IndexShard indexShard, Repository repository, SnapshotRecov
                 indexIdListener.onResponse(indexId);
             }
             assert indexShard.getEngineOrNull() == null;
-            indexIdListener.whenComplete(idx -> repository.restoreShard(indexShard.store(), restoreSource.snapshot().getSnapshotId(),
-                idx, snapshotShardId, indexShard.recoveryState(), restoreListener), restoreListener::onFailure);
-        } catch (Exception e) {
+            indexIdListener.whenComplete(
+                idx -> {
+                    if (IndexSettings.SKIP_FILES_RECOVERY_SETTING.get(indexShard.indexSettings().getSettings())) {
+                        logger.debug("[{}] skipping full restore from [{}] [{}]",
+                            shardId, restoreSource.snapshot().getRepository(), restoreSource.snapshot().getSnapshotId());
+                        restoreListener.onResponse(null);
+                    } else {
+                        repository.restoreShard(indexShard.store(), restoreSource.snapshot().getSnapshotId(),
+                            idx, snapshotShardId, indexShard.recoveryState(), restoreListener);
+                    }
+                }, restoreListener::onFailure);
+            } catch (Exception e) {
             restoreListener.onFailure(e);
         }
     }
@@ -506,4 +527,13 @@ private void bootstrap(final IndexShard indexShard, final Store store) throws IO
             indexShard.shardPath().resolveTranslog(), localCheckpoint, shardId, indexShard.getPendingPrimaryTerm());
         store.associateIndexWithNewTranslog(translogUUID);
     }
+
+    private void associateTranslogWithIndex(final IndexShard indexShard, final Store store) throws IOException {
+        assert IndexSettings.SKIP_FILES_RECOVERY_SETTING.get(indexShard.indexSettings().getSettings());
+        final SegmentInfos segmentInfos = store.readLastCommittedSegmentsInfo();
+        final long localCheckpoint = Long.parseLong(segmentInfos.userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY));
+        final long primaryTerm = indexShard.getPendingPrimaryTerm();
+        final String translogUUID = segmentInfos.userData.get(Translog.TRANSLOG_UUID_KEY);
+        Translog.createEmptyTranslog(indexShard.shardPath().resolveTranslog(), shardId, localCheckpoint, primaryTerm, translogUUID, null);
+    }
 }

server/src/main/java/org/elasticsearch/index/translog/Translog.java

@@ -23,6 +23,8 @@
 import org.apache.lucene.index.Term;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.Version;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -1838,20 +1840,41 @@ public static String createEmptyTranslog(final Path location, final long initial
 
     static String createEmptyTranslog(Path location, long initialGlobalCheckpoint, ShardId shardId,
                                       ChannelFactory channelFactory, long primaryTerm) throws IOException {
+        return createEmptyTranslog(location, shardId, initialGlobalCheckpoint, primaryTerm, null, channelFactory);
+    }
+
+    public static String createEmptyTranslog(final Path location,
+                                             final ShardId shardId,
+                                             final long initialGlobalCheckpoint,
+                                             final long primaryTerm,
+                                             @Nullable final String translogUUID,
+                                             @Nullable final ChannelFactory factory) throws IOException {
         IOUtils.rm(location);
         Files.createDirectories(location);
-        final Checkpoint checkpoint =
-            Checkpoint.emptyTranslogCheckpoint(0, 1, initialGlobalCheckpoint, 1);
+
+        final long generation = 1L;
+        final long minTranslogGeneration = 1L;
+        final ChannelFactory channelFactory = factory != null ? factory : FileChannel::open;
+        final String uuid = Strings.hasLength(translogUUID) ? translogUUID : UUIDs.randomBase64UUID();
         final Path checkpointFile = location.resolve(CHECKPOINT_FILE_NAME);
+        final Path translogFile = location.resolve(getFilename(generation));
+        final Checkpoint checkpoint = Checkpoint.emptyTranslogCheckpoint(0, generation, initialGlobalCheckpoint, minTranslogGeneration);
+
         Checkpoint.write(channelFactory, checkpointFile, checkpoint, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW);
         IOUtils.fsync(checkpointFile, false);
-        final String translogUUID = UUIDs.randomBase64UUID();
-        TranslogWriter writer = TranslogWriter.create(shardId, translogUUID, 1,
-            location.resolve(getFilename(1)), channelFactory,
-            new ByteSizeValue(10), 1, initialGlobalCheckpoint,
-            () -> { throw new UnsupportedOperationException(); }, () -> { throw new UnsupportedOperationException(); }, primaryTerm,
-                new TragicExceptionHolder(), seqNo -> { throw new UnsupportedOperationException(); });
+        final TranslogWriter writer = TranslogWriter.create(shardId, uuid, generation, translogFile, channelFactory,
+            new ByteSizeValue(10), minTranslogGeneration, initialGlobalCheckpoint,
+            () -> {
+                throw new UnsupportedOperationException();
+            }, () -> {
+                throw new UnsupportedOperationException();
+            },
+            primaryTerm,
+            new TragicExceptionHolder(),
+            seqNo -> {
+                throw new UnsupportedOperationException();
+            });
         writer.close();
-        return translogUUID;
+        return uuid;
     }
 }

server/src/main/java/org/elasticsearch/indices/recovery/PeerRecoveryTargetService.java

@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.lucene.index.SegmentInfos;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.RateLimiter;
 import org.elasticsearch.ElasticsearchException;
@@ -42,8 +43,10 @@
 import org.elasticsearch.common.util.CancellableThreads;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.engine.RecoveryEngineException;
 import org.elasticsearch.index.mapper.MapperException;
+import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.IllegalIndexShardStateException;
 import org.elasticsearch.index.shard.IndexEventListener;
 import org.elasticsearch.index.shard.IndexShard;
@@ -64,6 +67,7 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 
@@ -174,8 +178,19 @@ private void doRecovery(final long recoveryId) {
             try {
                 assert recoveryTarget.sourceNode() != null : "can not do a recovery without a source node";
                 logger.trace("{} preparing shard for peer recovery", recoveryTarget.shardId());
-                recoveryTarget.indexShard().prepareForIndexRecovery();
-                final long startingSeqNo = recoveryTarget.indexShard().recoverLocallyUpToGlobalCheckpoint();
+                final IndexShard indexShard = recoveryTarget.indexShard();
+                indexShard.prepareForIndexRecovery();
+                if (IndexSettings.SKIP_FILES_RECOVERY_SETTING.get(indexShard.indexSettings().getSettings())) {
+                    // associate a new empty translog with the last lucene commit, this way the next StartRecoveryRequest
+                    // will see shard files as if they were already on disk
+                    final SegmentInfos segmentInfos = indexShard.store().readLastCommittedSegmentsInfo();
+                    final long localCheckpoint = Long.parseLong(segmentInfos.userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY));
+                    final long primaryTerm = indexShard.getPendingPrimaryTerm();
+                    final String translogUUID = segmentInfos.userData.get(Translog.TRANSLOG_UUID_KEY);
+                    final Path location = indexShard.shardPath().resolveTranslog();
+                    Translog.createEmptyTranslog(location, indexShard.shardId(), localCheckpoint, primaryTerm, translogUUID, null);
+                }
+                final long startingSeqNo = indexShard.recoverLocallyUpToGlobalCheckpoint();
                 assert startingSeqNo == UNASSIGNED_SEQ_NO || recoveryTarget.state().getStage() == RecoveryState.Stage.TRANSLOG :
                     "unexpected recovery stage [" + recoveryTarget.state().getStage() + "] starting seqno [ " + startingSeqNo + "]";
                 request = getStartRecoveryRequest(logger, clusterService.localNode(), recoveryTarget, startingSeqNo);

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/SearchableSnapshotDirectory.java

@@ -5,9 +5,6 @@
  */
 package org.elasticsearch.index.store;
 
-import org.apache.lucene.index.IndexWriter;
-import org.apache.lucene.index.IndexWriterConfig;
-import org.apache.lucene.index.NoMergePolicy;
 import org.apache.lucene.store.BaseDirectory;
 import org.apache.lucene.store.BufferedIndexInput;
 import org.apache.lucene.store.Directory;
@@ -16,13 +13,10 @@
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.SingleInstanceLockFactory;
 import org.elasticsearch.common.blobstore.BlobContainer;
-import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.index.IndexSettings;
-import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
-import org.elasticsearch.index.translog.Translog;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.Repository;
@@ -35,8 +29,6 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.LongSupplier;
@@ -172,26 +164,6 @@ public static Directory create(RepositoriesService repositories,
             directory = new CacheDirectory(directory, cache, cacheDir, snapshotId, indexId, shardPath.getShardId(),
                 currentTimeNanosSupplier);
         }
-        directory = new InMemoryNoOpCommitDirectory(directory);
-
-        final IndexWriterConfig indexWriterConfig = new IndexWriterConfig(null)
-            .setSoftDeletesField(Lucene.SOFT_DELETES_FIELD)
-            .setMergePolicy(NoMergePolicy.INSTANCE);
-
-        try (IndexWriter indexWriter = new IndexWriter(directory, indexWriterConfig)) {
-            final Map<String, String> userData = new HashMap<>();
-            indexWriter.getLiveCommitData().forEach(e -> userData.put(e.getKey(), e.getValue()));
-
-            final String translogUUID = Translog.createEmptyTranslog(shardPath.resolveTranslog(),
-                Long.parseLong(userData.get(SequenceNumbers.LOCAL_CHECKPOINT_KEY)),
-                shardPath.getShardId(), 0L);
-
-            userData.put(Translog.TRANSLOG_UUID_KEY, translogUUID);
-            indexWriter.setLiveCommitData(userData.entrySet());
-            indexWriter.commit();
-        }
-
-        return directory;
+        return new InMemoryNoOpCommitDirectory(directory);
     }
-
 }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/TransportMountSearchableSnapshotAction.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.Repository;
@@ -95,6 +96,7 @@ private static Settings buildIndexSettings(String repoName, SnapshotId snapshotI
             .put(SearchableSnapshots.SNAPSHOT_SNAPSHOT_ID_SETTING.getKey(), snapshotId.getUUID())
             .put(SearchableSnapshots.SNAPSHOT_INDEX_ID_SETTING.getKey(), indexId.getId())
             .put(INDEX_STORE_TYPE_SETTING.getKey(), SearchableSnapshots.SNAPSHOT_DIRECTORY_FACTORY_KEY)
+            .put(IndexSettings.SKIP_FILES_RECOVERY_SETTING.getKey(), true)
             .put(IndexMetaData.SETTING_BLOCKS_WRITE, true)
             .put(ExistingShardsAllocator.EXISTING_SHARDS_ALLOCATOR_SETTING.getKey(), SearchableSnapshotAllocator.ALLOCATOR_NAME)
             .build();