Untangle Engine Constructor logic (#28245)

Currently we have a fairly complicated logic in the engine constructor logic to deal with all the 
various ways we want to mutate the lucene index and translog we're opening.

We can:
1) Create an empty index
2) Use the lucene but create a new translog
3) Use both
4) Force a new history uuid in all cases.

This leads complicated code flows which makes it harder and harder to make sure we cover all the 
corner cases. This PR tries to take another approach. Constructing an InternalEngine always opens 
things as they are and all needed modifications are done by static methods directly on the 
directory, one at a time.

Author: bleskes

docs/reference/indices/flush.asciidoc

@@ -93,12 +93,12 @@ which returns something similar to:
                {
                  "commit" : {
                    "id" : "3M3zkw2GHMo2Y4h4/KFKCg==",
-                   "generation" : 2,
+                   "generation" : 3,
                    "user_data" : {
                      "translog_uuid" : "hnOG3xFcTDeoI_kvvvOdNA",
                      "history_uuid" : "XP7KDJGiS1a2fHYiFL5TXQ",
                      "local_checkpoint" : "-1",
-                     "translog_generation" : "2",
+                     "translog_generation" : "3",
                      "max_seq_no" : "-1",
                      "sync_id" : "AVvFY-071siAOuFGEO9P", <1>
                      "max_unsafe_auto_id_timestamp" : "-1"

server/src/main/java/org/elasticsearch/index/engine/CombinedDeletionPolicy.java

@@ -46,16 +46,14 @@
 public final class CombinedDeletionPolicy extends IndexDeletionPolicy {
     private final Logger logger;
     private final TranslogDeletionPolicy translogDeletionPolicy;
-    private final EngineConfig.OpenMode openMode;
     private final LongSupplier globalCheckpointSupplier;
     private final IndexCommit startingCommit;
     private final ObjectIntHashMap<IndexCommit> snapshottedCommits; // Number of snapshots held against each commit point.
     private volatile IndexCommit safeCommit; // the most recent safe commit point - its max_seqno at most the persisted global checkpoint.
     private volatile IndexCommit lastCommit; // the most recent commit point
 
-    CombinedDeletionPolicy(EngineConfig.OpenMode openMode, Logger logger, TranslogDeletionPolicy translogDeletionPolicy,
+    CombinedDeletionPolicy(Logger logger, TranslogDeletionPolicy translogDeletionPolicy,
                            LongSupplier globalCheckpointSupplier, IndexCommit startingCommit) {
-        this.openMode = openMode;
         this.logger = logger;
         this.translogDeletionPolicy = translogDeletionPolicy;
         this.globalCheckpointSupplier = globalCheckpointSupplier;
@@ -65,25 +63,11 @@ public final class CombinedDeletionPolicy extends IndexDeletionPolicy {
 
     @Override
     public synchronized void onInit(List<? extends IndexCommit> commits) throws IOException {
-        switch (openMode) {
-            case CREATE_INDEX_AND_TRANSLOG:
-                assert startingCommit == null : "CREATE_INDEX_AND_TRANSLOG must not have starting commit; commit [" + startingCommit + "]";
-                break;
-            case OPEN_INDEX_CREATE_TRANSLOG:
-            case OPEN_INDEX_AND_TRANSLOG:
-                assert commits.isEmpty() == false : "index is opened, but we have no commits";
-                assert startingCommit != null && commits.contains(startingCommit) : "Starting commit not in the existing commit list; "
-                    + "startingCommit [" + startingCommit + "], commit list [" + commits + "]";
-                keepOnlyStartingCommitOnInit(commits);
-                // OPEN_INDEX_CREATE_TRANSLOG can open an index commit from other shard with a different translog history,
-                // We therefore should not use that index commit to update the translog deletion policy.
-                if (openMode == EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG) {
-                    updateTranslogDeletionPolicy();
-                }
-                break;
-            default:
-                throw new IllegalArgumentException("unknown openMode [" + openMode + "]");
-        }
+        assert commits.isEmpty() == false : "index is opened, but we have no commits";
+        assert startingCommit != null && commits.contains(startingCommit) : "Starting commit not in the existing commit list; "
+            + "startingCommit [" + startingCommit + "], commit list [" + commits + "]";
+        keepOnlyStartingCommitOnInit(commits);
+        updateTranslogDeletionPolicy();
     }
 
     /**

server/src/main/java/org/elasticsearch/index/engine/EngineConfig.java

@@ -75,7 +75,6 @@ public final class EngineConfig {
     private final List<ReferenceManager.RefreshListener> internalRefreshListener;
     @Nullable
     private final Sort indexSort;
-    private final boolean forceNewHistoryUUID;
     private final TranslogRecoveryRunner translogRecoveryRunner;
     @Nullable
     private final CircuitBreakerService circuitBreakerService;
@@ -113,24 +112,20 @@ public final class EngineConfig {
         Property.IndexScope, Property.Dynamic);
 
     private final TranslogConfig translogConfig;
-    private final OpenMode openMode;
 
     /**
      * Creates a new {@link org.elasticsearch.index.engine.EngineConfig}
      */
-    public EngineConfig(OpenMode openMode, ShardId shardId, String allocationId, ThreadPool threadPool,
+    public EngineConfig(ShardId shardId, String allocationId, ThreadPool threadPool,
                         IndexSettings indexSettings, Engine.Warmer warmer, Store store,
                         MergePolicy mergePolicy, Analyzer analyzer,
                         Similarity similarity, CodecService codecService, Engine.EventListener eventListener,
                         QueryCache queryCache, QueryCachingPolicy queryCachingPolicy,
-                        boolean forceNewHistoryUUID, TranslogConfig translogConfig, TimeValue flushMergesAfter,
+                        TranslogConfig translogConfig, TimeValue flushMergesAfter,
                         List<ReferenceManager.RefreshListener> externalRefreshListener,
                         List<ReferenceManager.RefreshListener> internalRefreshListener, Sort indexSort,
                         TranslogRecoveryRunner translogRecoveryRunner, CircuitBreakerService circuitBreakerService,
                         LongSupplier globalCheckpointSupplier) {
-        if (openMode == null) {
-            throw new IllegalArgumentException("openMode must not be null");
-        }
         this.shardId = shardId;
         this.allocationId = allocationId;
         this.indexSettings = indexSettings;
@@ -151,8 +146,6 @@ public EngineConfig(OpenMode openMode, ShardId shardId, String allocationId, Thr
         this.queryCachingPolicy = queryCachingPolicy;
         this.translogConfig = translogConfig;
         this.flushMergesAfter = flushMergesAfter;
-        this.openMode = openMode;
-        this.forceNewHistoryUUID = forceNewHistoryUUID;
         this.externalRefreshListener = externalRefreshListener;
         this.internalRefreshListener = internalRefreshListener;
         this.indexSort = indexSort;
@@ -315,22 +308,6 @@ public TranslogConfig getTranslogConfig() {
      */
     public TimeValue getFlushMergesAfter() { return flushMergesAfter; }
 
-    /**
-     * Returns the {@link OpenMode} for this engine config.
-     */
-    public OpenMode getOpenMode() {
-        return openMode;
-    }
-
-
-    /**
-     * Returns true if a new history uuid must be generated. If false, a new uuid will only be generated if no existing
-     * one is found.
-     */
-    public boolean getForceNewHistoryUUID() {
-        return forceNewHistoryUUID;
-    }
-
     @FunctionalInterface
     public interface TranslogRecoveryRunner {
         int run(Engine engine, Translog.Snapshot snapshot) throws IOException;
@@ -343,20 +320,6 @@ public TranslogRecoveryRunner getTranslogRecoveryRunner() {
         return translogRecoveryRunner;
     }
 
-    /**
-     * Engine open mode defines how the engine should be opened or in other words what the engine should expect
-     * to recover from. We either create a brand new engine with a new index and translog or we recover from an existing index.
-     * If the index exists we also have the ability open only the index and create a new transaction log which happens
-     * during remote recovery since we have already transferred the index files but the translog is replayed from remote. The last
-     * and safest option opens the lucene index as well as it's referenced transaction log for a translog recovery.
-     * See also {@link Engine#recoverFromTranslog()}
-     */
-    public enum OpenMode {
-        CREATE_INDEX_AND_TRANSLOG,
-        OPEN_INDEX_CREATE_TRANSLOG,
-        OPEN_INDEX_AND_TRANSLOG;
-    }
-
     /**
      * The refresh listeners to add to Lucene for externally visible refreshes
      */

server/src/main/java/org/elasticsearch/index/engine/EngineDiskUtils.java

@@ -0,0 +1,144 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.index.engine;
+
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexCommit;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.NoMergePolicy;
+import org.apache.lucene.store.Directory;
+import org.elasticsearch.Assertions;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.index.seqno.SequenceNumbers;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.translog.Translog;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * This class contains utility methods for mutating the shard lucene index and translog as a preparation to be opened.
+ */
+public abstract class EngineDiskUtils {
+
+    /**
+     * creates an empty lucene index and a corresponding empty translog. Any existing data will be deleted.
+     */
+    public static void createEmpty(final Directory dir, final Path translogPath, final ShardId shardId) throws IOException {
+        try (IndexWriter writer = newIndexWriter(true, dir)) {
+            final String translogUuid = Translog.createEmptyTranslog(translogPath, SequenceNumbers.NO_OPS_PERFORMED, shardId);
+            final Map<String, String> map = new HashMap<>();
+            map.put(Translog.TRANSLOG_GENERATION_KEY, "1");
+            map.put(Translog.TRANSLOG_UUID_KEY, translogUuid);
+            map.put(Engine.HISTORY_UUID_KEY, UUIDs.randomBase64UUID());
+            map.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(SequenceNumbers.NO_OPS_PERFORMED));
+            map.put(SequenceNumbers.MAX_SEQ_NO, Long.toString(SequenceNumbers.NO_OPS_PERFORMED));
+            map.put(InternalEngine.MAX_UNSAFE_AUTO_ID_TIMESTAMP_COMMIT_ID, "-1");
+            updateCommitData(writer, map);
+        }
+    }
+
+
+    /**
+     * Converts an existing lucene index and marks it with a new history uuid. Also creates a new empty translog file.
+     * This is used to make sure no existing shard will recovery from this index using ops based recovery.
+     */
+    public static void bootstrapNewHistoryFromLuceneIndex(final Directory dir, final Path translogPath, final ShardId shardId)
+        throws IOException {
+        try (IndexWriter writer = newIndexWriter(false, dir)) {
+            final Map<String, String> userData = getUserData(writer);
+            final long maxSeqNo = Long.parseLong(userData.get(SequenceNumbers.MAX_SEQ_NO));
+            final String translogUuid = Translog.createEmptyTranslog(translogPath, maxSeqNo, shardId);
+            final Map<String, String> map = new HashMap<>();
+            map.put(Translog.TRANSLOG_GENERATION_KEY, "1");
+            map.put(Translog.TRANSLOG_UUID_KEY, translogUuid);
+            map.put(Engine.HISTORY_UUID_KEY, UUIDs.randomBase64UUID());
+            map.put(SequenceNumbers.LOCAL_CHECKPOINT_KEY, Long.toString(maxSeqNo));
+            updateCommitData(writer, map);
+        }
+    }
+
+    /**
+     * Creates a new empty translog and associates it with an existing lucene index.
+     */
+    public static void createNewTranslog(final Directory dir, final Path translogPath, long initialGlobalCheckpoint, final ShardId shardId)
+        throws IOException {
+        if (Assertions.ENABLED) {
+            final List<IndexCommit> existingCommits = DirectoryReader.listCommits(dir);
+            assert existingCommits.size() == 1 : "creating a translog translog should have one commit, commits[" + existingCommits + "]";
+            SequenceNumbers.CommitInfo commitInfo = Store.loadSeqNoInfo(existingCommits.get(0));
+            assert commitInfo.localCheckpoint >= initialGlobalCheckpoint :
+                "trying to create a shard whose local checkpoint [" + commitInfo.localCheckpoint + "] is < global checkpoint ["
+                + initialGlobalCheckpoint + "]";
+        }
+
+        try (IndexWriter writer = newIndexWriter(false, dir)) {
+            final String translogUuid = Translog.createEmptyTranslog(translogPath, initialGlobalCheckpoint, shardId);
+            final Map<String, String> map = new HashMap<>();
+            map.put(Translog.TRANSLOG_GENERATION_KEY, "1");
+            map.put(Translog.TRANSLOG_UUID_KEY, translogUuid);
+            updateCommitData(writer, map);
+        }
+    }
+
+
+    /**
+     * Checks that the Lucene index contains a history uuid marker. If not, a new one is generated and committed.
+     */
+    public static void ensureIndexHasHistoryUUID(final Directory dir) throws IOException {
+        try (IndexWriter writer = newIndexWriter(false, dir)) {
+            final Map<String, String> userData = getUserData(writer);
+            if (userData.containsKey(Engine.HISTORY_UUID_KEY) == false) {
+                updateCommitData(writer, Collections.singletonMap(Engine.HISTORY_UUID_KEY, UUIDs.randomBase64UUID()));
+            }
+        }
+    }
+
+    private static void updateCommitData(IndexWriter writer, Map<String, String> keysToUpdate) throws IOException {
+        final Map<String, String> userData = getUserData(writer);
+        userData.putAll(keysToUpdate);
+        writer.setLiveCommitData(userData.entrySet());
+        writer.commit();
+    }
+
+    private static Map<String, String> getUserData(IndexWriter writer) {
+        final Map<String, String> userData = new HashMap<>();
+        writer.getLiveCommitData().forEach(e -> userData.put(e.getKey(), e.getValue()));
+        return userData;
+    }
+
+    private static IndexWriter newIndexWriter(final boolean create, final Directory dir) throws IOException {
+        IndexWriterConfig iwc = new IndexWriterConfig(null)
+            .setCommitOnClose(false)
+            // we don't want merges to happen here - we call maybe merge on the engine
+            // later once we stared it up otherwise we would need to wait for it here
+            // we also don't specify a codec here and merges should use the engines for this index
+            .setMergePolicy(NoMergePolicy.INSTANCE)
+            .setOpenMode(create ? IndexWriterConfig.OpenMode.CREATE : IndexWriterConfig.OpenMode.APPEND);
+        return new IndexWriter(dir, iwc);
+    }
+}