Changes for remote translog backed by blob store

Signed-off-by: Bukhtawar Khan <bukhtawa@amazon.com>

server/src/main/java/org/opensearch/index/IndexModule.java

@@ -77,6 +77,7 @@
 import org.opensearch.indices.mapper.MapperRegistry;
 import org.opensearch.indices.recovery.RecoveryState;
 import org.opensearch.plugins.IndexStorePlugin;
+import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.search.aggregations.support.ValuesSourceRegistry;
 import org.opensearch.threadpool.ThreadPool;
@@ -91,10 +92,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.BiFunction;
-import java.util.function.BooleanSupplier;
-import java.util.function.Consumer;
-import java.util.function.Function;
+import java.util.function.*;
 
 /**
  * IndexModule represents the central extension point for index level custom implementations like:
@@ -475,7 +473,8 @@ public IndexService newIndexService(
         NamedWriteableRegistry namedWriteableRegistry,
         BooleanSupplier idFieldDataEnabled,
         ValuesSourceRegistry valuesSourceRegistry,
-        RemoteDirectoryFactory remoteDirectoryFactory
+        RemoteDirectoryFactory remoteDirectoryFactory,
+        Supplier<RepositoriesService> repositoriesServiceSupplier
     ) throws IOException {
         final IndexEventListener eventListener = freeze();
         Function<IndexService, CheckedFunction<DirectoryReader, DirectoryReader, IOException>> readerWrapperFactory = indexReaderWrapper
@@ -530,7 +529,8 @@ public IndexService newIndexService(
                 allowExpensiveQueries,
                 expressionResolver,
                 valuesSourceRegistry,
-                recoveryStateFactory
+                recoveryStateFactory,
+                repositoriesServiceSupplier
             );
             success = true;
             return indexService;

server/src/main/java/org/opensearch/index/IndexService.java

@@ -88,6 +88,8 @@
 import org.opensearch.index.shard.ShardPath;
 import org.opensearch.index.similarity.SimilarityService;
 import org.opensearch.index.store.Store;
+import org.opensearch.index.translog.InternalTranslogFactory;
+import org.opensearch.index.translog.RemoteBlobStoreInternalTranslogFactory;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.indices.cluster.IndicesClusterStateService;
@@ -96,6 +98,7 @@
 import org.opensearch.indices.recovery.RecoveryState;
 import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
 import org.opensearch.plugins.IndexStorePlugin;
+import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.search.aggregations.support.ValuesSourceRegistry;
 import org.opensearch.threadpool.ThreadPool;
@@ -172,6 +175,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final IndexNameExpressionResolver expressionResolver;
     private final Supplier<Sort> indexSortSupplier;
     private final ValuesSourceRegistry valuesSourceRegistry;
+    private final Supplier<RepositoriesService> repositoriesServiceSupplier;
 
     public IndexService(
         IndexSettings indexSettings,
@@ -203,7 +207,8 @@ public IndexService(
         BooleanSupplier allowExpensiveQueries,
         IndexNameExpressionResolver expressionResolver,
         ValuesSourceRegistry valuesSourceRegistry,
-        IndexStorePlugin.RecoveryStateFactory recoveryStateFactory
+        IndexStorePlugin.RecoveryStateFactory recoveryStateFactory,
+        Supplier<RepositoriesService> repositoriesServiceSupplier
     ) {
         super(indexSettings);
         this.allowExpensiveQueries = allowExpensiveQueries;
@@ -275,6 +280,7 @@ public IndexService(
         this.trimTranslogTask = new AsyncTrimTranslogTask(this);
         this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this);
         this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
+        this.repositoriesServiceSupplier = repositoriesServiceSupplier;
         updateFsyncTaskIfNecessary();
     }
 
@@ -553,6 +559,8 @@ public synchronized IndexShard createShard(
                 () -> globalCheckpointSyncer.accept(shardId),
                 retentionLeaseSyncer,
                 circuitBreakerService,
+                this.indexSettings.isSegRepEnabled() && this.indexSettings.isRemoteStoreEnabled() ? new InternalTranslogFactory() :
+                    new RemoteBlobStoreInternalTranslogFactory(repositoriesServiceSupplier, clusterService.state().metadata().clusterUUID()),
                 this.indexSettings.isSegRepEnabled() && routing.primary() ? checkpointPublisher : null,
                 remoteStore
             );

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

@@ -31,6 +31,7 @@
 
 package org.opensearch.index.engine;
 
+import com.sun.jna.StringArray;
 import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.codecs.Codec;
 import org.apache.lucene.index.MergePolicy;
@@ -51,8 +52,10 @@
 import org.opensearch.index.seqno.RetentionLeases;
 import org.opensearch.index.shard.ShardId;
 import org.opensearch.index.store.Store;
+import org.opensearch.index.translog.InternalTranslogFactory;
 import org.opensearch.index.translog.TranslogConfig;
 import org.opensearch.index.translog.TranslogDeletionPolicyFactory;
+import org.opensearch.index.translog.TranslogFactory;
 import org.opensearch.indices.IndexingMemoryController;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.threadpool.ThreadPool;
@@ -148,6 +151,8 @@ public Supplier<RetentionLeases> retentionLeasesSupplier() {
         Property.Dynamic
     );
 
+    private final TranslogFactory translogFactory;
+
     private final TranslogConfig translogConfig;
 
     public EngineConfig(
@@ -253,7 +258,8 @@ public EngineConfig(
             retentionLeasesSupplier,
             primaryTermSupplier,
             tombstoneDocSupplier,
-            false
+            false,
+            new InternalTranslogFactory()
         );
     }
 
@@ -284,7 +290,8 @@ public EngineConfig(
         Supplier<RetentionLeases> retentionLeasesSupplier,
         LongSupplier primaryTermSupplier,
         TombstoneDocSupplier tombstoneDocSupplier,
-        boolean isReadOnlyReplica
+        boolean isReadOnlyReplica,
+        TranslogFactory translogFactory
     ) {
         if (isReadOnlyReplica && indexSettings.isSegRepEnabled() == false) {
             throw new IllegalArgumentException("Shard can only be wired as a read only replica with Segment Replication enabled");
@@ -328,6 +335,7 @@ public EngineConfig(
         this.primaryTermSupplier = primaryTermSupplier;
         this.tombstoneDocSupplier = tombstoneDocSupplier;
         this.isReadOnlyReplica = isReadOnlyReplica;
+        this.translogFactory = translogFactory;
     }
 
     /**
@@ -532,6 +540,14 @@ public boolean isReadOnlyReplica() {
         return indexSettings.isSegRepEnabled() && isReadOnlyReplica;
     }
 
+    /**
+     * Returns the underlying translog factory
+     * @return the translog factory
+     */
+    public TranslogFactory getTranslogFactory() {
+        return translogFactory;
+    }
+
     /**
      * A supplier supplies tombstone documents which will be used in soft-update methods.
      * The returned document consists only _uid, _seqno, _term and _version fields; other metadata fields are excluded.

server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java

@@ -28,6 +28,7 @@
 import org.opensearch.index.store.Store;
 import org.opensearch.index.translog.TranslogConfig;
 import org.opensearch.index.translog.TranslogDeletionPolicyFactory;
+import org.opensearch.index.translog.TranslogFactory;
 import org.opensearch.indices.breaker.CircuitBreakerService;
 import org.opensearch.plugins.EnginePlugin;
 import org.opensearch.plugins.PluginsService;
@@ -147,7 +148,8 @@ public EngineConfig newEngineConfig(
         Supplier<RetentionLeases> retentionLeasesSupplier,
         LongSupplier primaryTermSupplier,
         EngineConfig.TombstoneDocSupplier tombstoneDocSupplier,
-        boolean isReadOnlyReplica
+        boolean isReadOnlyReplica,
+        TranslogFactory translogFactory
     ) {
         CodecService codecServiceToUse = codecService;
         if (codecService == null && this.codecServiceFactory != null) {
@@ -178,7 +180,8 @@ public EngineConfig newEngineConfig(
             retentionLeasesSupplier,
             primaryTermSupplier,
             tombstoneDocSupplier,
-            isReadOnlyReplica
+            isReadOnlyReplica,
+            translogFactory
         );
     }
 

server/src/main/java/org/opensearch/index/engine/EngineFactory.java

@@ -31,6 +31,8 @@
 
 package org.opensearch.index.engine;
 
+import org.opensearch.index.translog.TranslogFactory;
+
 /**
  * Simple Engine Factory
  *

server/src/main/java/org/opensearch/index/engine/InternalEngine.java

@@ -104,12 +104,7 @@
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.shard.OpenSearchMergePolicy;
 import org.opensearch.index.shard.ShardId;
-import org.opensearch.index.translog.Translog;
-import org.opensearch.index.translog.TranslogCorruptedException;
-import org.opensearch.index.translog.TranslogDeletionPolicy;
-import org.opensearch.index.translog.TranslogManager;
-import org.opensearch.index.translog.TranslogException;
-import org.opensearch.index.translog.InternalTranslogManager;
+import org.opensearch.index.translog.*;
 import org.opensearch.index.translog.listener.TranslogEventListener;
 import org.opensearch.index.translog.listener.CompositeTranslogEventListener;
 import org.opensearch.search.suggest.completion.CompletionStats;
@@ -289,7 +284,8 @@ public void onFailure(String reason, Exception ex) {
                     () -> getLocalCheckpointTracker(),
                     translogUUID,
                     new CompositeTranslogEventListener(Arrays.asList(internalTranslogEventListener, translogEventListener), shardId),
-                    this::ensureOpen
+                    this::ensureOpen,
+                    config().getTranslogFactory()
                 );
                 this.translogManager = translogManagerRef;
                 this.softDeletesPolicy = newSoftDeletesPolicy();

server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java

@@ -95,7 +95,8 @@ public void onAfterTranslogSync() {
                         }
                     }
                 },
-                this
+                this,
+                config().getTranslogFactory()
             );
             this.translogManager = translogManagerRef;
         } catch (IOException e) {

server/src/main/java/org/opensearch/index/engine/NoOpEngine.java

@@ -44,13 +44,7 @@
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.shard.DocsStats;
 import org.opensearch.index.store.Store;
-import org.opensearch.index.translog.Translog;
-import org.opensearch.index.translog.TranslogManager;
-import org.opensearch.index.translog.TranslogConfig;
-import org.opensearch.index.translog.TranslogException;
-import org.opensearch.index.translog.NoOpTranslogManager;
-import org.opensearch.index.translog.DefaultTranslogDeletionPolicy;
-import org.opensearch.index.translog.TranslogDeletionPolicy;
+import org.opensearch.index.translog.*;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -195,7 +189,7 @@ public void trimUnreferencedTranslogFiles() throws TranslogException {
                             final TranslogDeletionPolicy translogDeletionPolicy = new DefaultTranslogDeletionPolicy(-1, -1, 0);
                             translogDeletionPolicy.setLocalCheckpointOfSafeCommit(localCheckpoint);
                             try (
-                                Translog translog = new Translog(
+                                Translog translog = engineConfig.getTranslogFactory().newTranslog(
                                     translogConfig,
                                     translogUuid,
                                     translogDeletionPolicy,

server/src/main/java/org/opensearch/index/engine/ReadOnlyEngine.java

@@ -258,7 +258,7 @@ private static TranslogStats translogStats(final EngineConfig config, final Segm
         final long localCheckpoint = Long.parseLong(infos.getUserData().get(SequenceNumbers.LOCAL_CHECKPOINT_KEY));
         translogDeletionPolicy.setLocalCheckpointOfSafeCommit(localCheckpoint);
         try (
-            Translog translog = new Translog(
+            Translog translog = config.getTranslogFactory().newTranslog(
                 translogConfig,
                 translogUuid,
                 translogDeletionPolicy,

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -149,10 +149,7 @@
 import org.opensearch.index.store.Store.MetadataSnapshot;
 import org.opensearch.index.store.StoreFileMetadata;
 import org.opensearch.index.store.StoreStats;
-import org.opensearch.index.translog.Translog;
-import org.opensearch.index.translog.TranslogConfig;
-import org.opensearch.index.translog.TranslogRecoveryRunner;
-import org.opensearch.index.translog.TranslogStats;
+import org.opensearch.index.translog.*;
 import org.opensearch.index.warmer.ShardIndexWarmerService;
 import org.opensearch.index.warmer.WarmerStats;
 import org.opensearch.indices.IndexingMemoryController;
@@ -308,6 +305,7 @@ Runnable getGlobalCheckpointSyncer() {
     private final ReferenceManager.RefreshListener checkpointRefreshListener;
 
     private final Store remoteStore;
+    private final TranslogFactory translogFactory;
 
     public IndexShard(
         final ShardRouting shardRouting,
@@ -330,6 +328,7 @@ public IndexShard(
         final Runnable globalCheckpointSyncer,
         final RetentionLeaseSyncer retentionLeaseSyncer,
         final CircuitBreakerService circuitBreakerService,
+        final TranslogFactory translogFactory,
         @Nullable final SegmentReplicationCheckpointPublisher checkpointPublisher,
         @Nullable final Store remoteStore
     ) throws IOException {
@@ -420,6 +419,7 @@ public boolean shouldCache(Query query) {
             this.checkpointRefreshListener = null;
         }
         this.remoteStore = remoteStore;
+        this.translogFactory = translogFactory;
     }
 
     public ThreadPool getThreadPool() {
@@ -3242,7 +3242,8 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
             replicationTracker::getRetentionLeases,
             () -> getOperationPrimaryTerm(),
             tombstoneDocSupplier(),
-            indexSettings.isSegRepEnabled() && shardRouting.primary() == false
+            indexSettings.isSegRepEnabled() && shardRouting.primary() == false,
+            translogFactory
         );
     }
 

server/src/main/java/org/opensearch/index/translog/InternalTranslogFactory.java

@@ -0,0 +1,34 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.translog;
+
+import java.io.IOException;
+import java.util.function.LongConsumer;
+import java.util.function.LongSupplier;
+
+public class InternalTranslogFactory implements TranslogFactory {
+
+    @Override
+    public Translog newTranslog(TranslogConfig translogConfig,
+                                String translogUUID,
+                                TranslogDeletionPolicy translogDeletionPolicy,
+                                LongSupplier globalCheckpointSupplier,
+                                LongSupplier primaryTermSupplier,
+                                LongConsumer persistedSequenceNumberConsumer) throws IOException {
+
+        return new LocalTranslog(
+            translogConfig,
+            translogUUID,
+            translogDeletionPolicy,
+            globalCheckpointSupplier,
+            primaryTermSupplier,
+            persistedSequenceNumberConsumer
+        );
+    }
+}

server/src/main/java/org/opensearch/index/translog/InternalTranslogManager.java

@@ -54,7 +54,8 @@ public InternalTranslogManager(
         Supplier<LocalCheckpointTracker> localCheckpointTrackerSupplier,
         String translogUUID,
         TranslogEventListener translogEventListener,
-        LifecycleAware engineLifeCycleAware
+        LifecycleAware engineLifeCycleAware,
+        TranslogFactory translogFactory
     ) throws IOException {
         this.shardId = shardId;
         this.readLock = readLock;
@@ -67,7 +68,7 @@ public InternalTranslogManager(
             if (tracker != null) {
                 tracker.markSeqNoAsPersisted(seqNo);
             }
-        }, translogUUID);
+        }, translogUUID, translogFactory);
         assert translog.getGeneration() != null;
         this.translog = translog;
         assert pendingTranslogRecovery.get() == false : "translog recovery can't be pending before we set it";
@@ -333,10 +334,11 @@ protected Translog openTranslog(
         TranslogDeletionPolicy translogDeletionPolicy,
         LongSupplier globalCheckpointSupplier,
         LongConsumer persistedSequenceNumberConsumer,
-        String translogUUID
+        String translogUUID,
+        TranslogFactory translogFactory
     ) throws IOException {
 
-        return new Translog(
+        return translogFactory.newTranslog(
             translogConfig,
             translogUUID,
             translogDeletionPolicy,