Upload translog only if primaryMode is true

Signed-off-by: Ashish Singh <ssashish@amazon.com>
opensearch-project · Jan 10, 2023 · 1f7afe8 · 1f7afe8
1 parent ff8a3af
commit 1f7afe8
Show file tree

Hide file tree

Showing 16 changed files with 87 additions and 28 deletions.
diff --git a/server/src/main/java/org/opensearch/index/engine/EngineConfig.java b/server/src/main/java/org/opensearch/index/engine/EngineConfig.java
@@ -61,6 +61,7 @@
 
 import java.util.List;
 import java.util.Objects;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -100,6 +101,7 @@ public final class EngineConfig {
     private final LongSupplier globalCheckpointSupplier;
     private final Supplier<RetentionLeases> retentionLeasesSupplier;
     private final boolean isReadOnlyReplica;
+    private final BooleanSupplier primaryModeSupplier;
 
     /**
      * A supplier of the outstanding retention leases. This is used during merged operations to determine which operations that have been
@@ -200,6 +202,7 @@ private EngineConfig(Builder builder) {
         this.primaryTermSupplier = builder.primaryTermSupplier;
         this.tombstoneDocSupplier = builder.tombstoneDocSupplier;
         this.isReadOnlyReplica = builder.isReadOnlyReplica;
+        this.primaryModeSupplier = builder.primaryModeSupplier;
         this.translogFactory = builder.translogFactory;
     }
 
@@ -405,6 +408,14 @@ public boolean isReadOnlyReplica() {
         return indexSettings.isSegRepEnabled() && isReadOnlyReplica;
     }
 
+    /**
+     * Returns the underlying primaryModeSupplier.
+     * @return the primary mode supplier.
+     */
+    public BooleanSupplier getPrimaryModeSupplier() {
+        return primaryModeSupplier;
+    }
+
     /**
      * Returns the underlying translog factory
      * @return the translog factory
@@ -470,6 +481,7 @@ public static class Builder {
         private TombstoneDocSupplier tombstoneDocSupplier;
         private TranslogDeletionPolicyFactory translogDeletionPolicyFactory;
         private boolean isReadOnlyReplica;
+        private BooleanSupplier primaryModeSupplier;
         private TranslogFactory translogFactory = new InternalTranslogFactory();
 
         public Builder shardId(ShardId shardId) {
@@ -592,6 +604,11 @@ public Builder readOnlyReplica(boolean isReadOnlyReplica) {
             return this;
         }
 
+        public Builder primaryModeSupplier(BooleanSupplier primaryModeSupplier) {
+            this.primaryModeSupplier = primaryModeSupplier;
+            return this;
+        }
+
         public Builder translogFactory(TranslogFactory translogFactory) {
             this.translogFactory = translogFactory;
             return this;

diff --git a/server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java b/server/src/main/java/org/opensearch/index/engine/EngineConfigFactory.java
@@ -38,6 +38,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -149,6 +150,7 @@ public EngineConfig newEngineConfig(
         LongSupplier primaryTermSupplier,
         EngineConfig.TombstoneDocSupplier tombstoneDocSupplier,
         boolean isReadOnlyReplica,
+        BooleanSupplier primaryModeSupplier,
         TranslogFactory translogFactory
     ) {
         CodecService codecServiceToUse = codecService;
@@ -180,6 +182,7 @@ public EngineConfig newEngineConfig(
             .primaryTermSupplier(primaryTermSupplier)
             .tombstoneDocSupplier(tombstoneDocSupplier)
             .readOnlyReplica(isReadOnlyReplica)
+            .primaryModeSupplier(primaryModeSupplier)
             .translogFactory(translogFactory)
             .build();
     }

diff --git a/server/src/main/java/org/opensearch/index/engine/InternalEngine.java b/server/src/main/java/org/opensearch/index/engine/InternalEngine.java
@@ -286,11 +286,12 @@ public void onFailure(String reason, Exception ex) {
                     translogDeletionPolicy,
                     shardId,
                     readLock,
-                    () -> getLocalCheckpointTracker(),
+                    this::getLocalCheckpointTracker,
                     translogUUID,
                     new CompositeTranslogEventListener(Arrays.asList(internalTranslogEventListener, translogEventListener), shardId),
-                    this::ensureOpen,
-                    engineConfig.getTranslogFactory()
+                    this,
+                    engineConfig.getTranslogFactory(),
+                    engineConfig.getPrimaryModeSupplier()
                 );
                 this.translogManager = translogManagerRef;
                 this.softDeletesPolicy = newSoftDeletesPolicy();

diff --git a/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java b/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java
@@ -107,7 +107,8 @@ public void onAfterTranslogSync() {
                     }
                 },
                 this,
-                engineConfig.getTranslogFactory()
+                engineConfig.getTranslogFactory(),
+                engineConfig.getPrimaryModeSupplier()
             );
             this.translogManager = translogManagerRef;
         } catch (IOException e) {

diff --git a/server/src/main/java/org/opensearch/index/engine/NoOpEngine.java b/server/src/main/java/org/opensearch/index/engine/NoOpEngine.java
@@ -202,7 +202,8 @@ public void trimUnreferencedTranslogFiles() throws TranslogException {
                                         translogDeletionPolicy,
                                         engineConfig.getGlobalCheckpointSupplier(),
                                         engineConfig.getPrimaryTermSupplier(),
-                                        seqNo -> {}
+                                        seqNo -> {},
+                                        engineConfig.getPrimaryModeSupplier()
                                     )
                             ) {
                                 translog.trimUnreferencedReaders();

diff --git a/server/src/main/java/org/opensearch/index/engine/ReadOnlyEngine.java b/server/src/main/java/org/opensearch/index/engine/ReadOnlyEngine.java
@@ -259,7 +259,8 @@ private static TranslogStats translogStats(final EngineConfig config, final Segm
                     translogDeletionPolicy,
                     config.getGlobalCheckpointSupplier(),
                     config.getPrimaryTermSupplier(),
-                    seqNo -> {}
+                    seqNo -> {},
+                    config.getPrimaryModeSupplier()
                 )
         ) {
             return translog.stats();

diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -3421,6 +3421,7 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
             () -> getOperationPrimaryTerm(),
             tombstoneDocSupplier(),
             isReadOnlyReplica,
+            replicationTracker::isPrimaryMode,
             translogFactorySupplier.apply(indexSettings, shardRouting)
         );
     }

diff --git a/server/src/main/java/org/opensearch/index/translog/InternalTranslogFactory.java b/server/src/main/java/org/opensearch/index/translog/InternalTranslogFactory.java
@@ -9,6 +9,7 @@
 package org.opensearch.index.translog;
 
 import java.io.IOException;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 
@@ -26,7 +27,8 @@ public Translog newTranslog(
         TranslogDeletionPolicy translogDeletionPolicy,
         LongSupplier globalCheckpointSupplier,
         LongSupplier primaryTermSupplier,
-        LongConsumer persistedSequenceNumberConsumer
+        LongConsumer persistedSequenceNumberConsumer,
+        BooleanSupplier primaryModeSupplier
     ) throws IOException {
 
         return new LocalTranslog(

diff --git a/server/src/main/java/org/opensearch/index/translog/InternalTranslogManager.java b/server/src/main/java/org/opensearch/index/translog/InternalTranslogManager.java
@@ -22,6 +22,7 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -55,7 +56,8 @@ public InternalTranslogManager(
         String translogUUID,
         TranslogEventListener translogEventListener,
         LifecycleAware engineLifeCycleAware,
-        TranslogFactory translogFactory
+        TranslogFactory translogFactory,
+        BooleanSupplier primaryModeSupplier
     ) throws IOException {
         this.shardId = shardId;
         this.readLock = readLock;
@@ -68,7 +70,7 @@ public InternalTranslogManager(
             if (tracker != null) {
                 tracker.markSeqNoAsPersisted(seqNo);
             }
-        }, translogUUID, translogFactory);
+        }, translogUUID, translogFactory, primaryModeSupplier);
         assert translog.getGeneration() != null;
         this.translog = translog;
         assert pendingTranslogRecovery.get() == false : "translog recovery can't be pending before we set it";
@@ -345,16 +347,17 @@ protected Translog openTranslog(
         LongSupplier globalCheckpointSupplier,
         LongConsumer persistedSequenceNumberConsumer,
         String translogUUID,
-        TranslogFactory translogFactory
+        TranslogFactory translogFactory,
+        BooleanSupplier primaryModeSupplier
     ) throws IOException {
-
         return translogFactory.newTranslog(
             translogConfig,
             translogUUID,
             translogDeletionPolicy,
             globalCheckpointSupplier,
             primaryTermSupplier,
-            persistedSequenceNumberConsumer
+            persistedSequenceNumberConsumer,
+            primaryModeSupplier
         );
     }
 

diff --git a/...r/src/main/java/org/opensearch/index/translog/RemoteBlobStoreInternalTranslogFactory.java b/...r/src/main/java/org/opensearch/index/translog/RemoteBlobStoreInternalTranslogFactory.java
@@ -16,6 +16,7 @@
 
 import java.io.IOException;
 import java.util.concurrent.ExecutorService;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
@@ -53,7 +54,8 @@ public Translog newTranslog(
         TranslogDeletionPolicy deletionPolicy,
         LongSupplier globalCheckpointSupplier,
         LongSupplier primaryTermSupplier,
-        LongConsumer persistedSequenceNumberConsumer
+        LongConsumer persistedSequenceNumberConsumer,
+        BooleanSupplier primaryModeSupplier
     ) throws IOException {
 
         assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
@@ -66,7 +68,8 @@ public Translog newTranslog(
             primaryTermSupplier,
             persistedSequenceNumberConsumer,
             blobStoreRepository,
-            executorService
+            executorService,
+            primaryModeSupplier
         );
     }
 }
diff --git a/server/src/main/java/org/opensearch/index/translog/RemoteFsTranslog.java b/server/src/main/java/org/opensearch/index/translog/RemoteFsTranslog.java
@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 
@@ -45,6 +46,7 @@ public class RemoteFsTranslog extends Translog {
     private final BlobStoreRepository blobStoreRepository;
     private final TranslogTransferManager translogTransferManager;
     private final FileTransferTracker fileTransferTracker;
+    private final BooleanSupplier primaryModeSupplier;
     private volatile long maxRemoteTranslogGenerationUploaded;
 
     private volatile long minSeqNoToKeep;
@@ -57,10 +59,12 @@ public RemoteFsTranslog(
         LongSupplier primaryTermSupplier,
         LongConsumer persistedSequenceNumberConsumer,
         BlobStoreRepository blobStoreRepository,
-        ExecutorService executorService
+        ExecutorService executorService,
+        BooleanSupplier primaryModeSupplier
     ) throws IOException {
         super(config, translogUUID, deletionPolicy, globalCheckpointSupplier, primaryTermSupplier, persistedSequenceNumberConsumer);
         this.blobStoreRepository = blobStoreRepository;
+        this.primaryModeSupplier = primaryModeSupplier;
         fileTransferTracker = new FileTransferTracker(shardId);
         this.translogTransferManager = buildTranslogTransferManager(blobStoreRepository, executorService, shardId, fileTransferTracker);
 
@@ -198,7 +202,13 @@ private boolean prepareAndUpload(Long primaryTerm, Long generation) throws IOExc
     }
 
     private boolean upload(Long primaryTerm, Long generation) throws IOException {
-        logger.trace("uploading translog for {} {} ", primaryTerm, generation);
+        boolean primaryMode = primaryModeSupplier.getAsBoolean();
+        if (primaryMode) {
+            logger.trace("skipped uploading translog for {} {}", primaryTerm, generation);
+            // NO-OP
+            return true;
+        }
+        logger.trace("uploading translog for {} {}", primaryTerm, generation);
         try (
             TranslogCheckpointTransferSnapshot transferSnapshotProvider = new TranslogCheckpointTransferSnapshot.Builder(
                 primaryTerm,

diff --git a/server/src/main/java/org/opensearch/index/translog/TranslogFactory.java b/server/src/main/java/org/opensearch/index/translog/TranslogFactory.java
@@ -9,6 +9,7 @@
 package org.opensearch.index.translog;
 
 import java.io.IOException;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
 
@@ -27,6 +28,7 @@ Translog newTranslog(
         final TranslogDeletionPolicy deletionPolicy,
         final LongSupplier globalCheckpointSupplier,
         final LongSupplier primaryTermSupplier,
-        final LongConsumer persistedSequenceNumberConsumer
+        final LongConsumer persistedSequenceNumberConsumer,
+        final BooleanSupplier primaryModeSupplier
     ) throws IOException;
 }
diff --git a/server/src/main/java/org/opensearch/index/translog/WriteOnlyTranslogManager.java b/server/src/main/java/org/opensearch/index/translog/WriteOnlyTranslogManager.java
@@ -15,6 +15,7 @@
 import org.opensearch.index.translog.listener.TranslogEventListener;
 
 import java.io.IOException;
+import java.util.function.BooleanSupplier;
 import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
@@ -36,7 +37,8 @@ public WriteOnlyTranslogManager(
         String translogUUID,
         TranslogEventListener translogEventListener,
         LifecycleAware engineLifecycleAware,
-        TranslogFactory translogFactory
+        TranslogFactory translogFactory,
+        BooleanSupplier primaryModeSupplier
     ) throws IOException {
         super(
             translogConfig,
@@ -49,7 +51,8 @@ public WriteOnlyTranslogManager(
             translogUUID,
             translogEventListener,
             engineLifecycleAware,
-            translogFactory
+            translogFactory,
+            primaryModeSupplier
         );
     }
 

diff --git a/server/src/test/java/org/opensearch/index/engine/EngineConfigFactoryTests.java b/server/src/test/java/org/opensearch/index/engine/EngineConfigFactoryTests.java
@@ -68,6 +68,7 @@ public void testCreateEngineConfigFromFactory() {
             null,
             null,
             false,
+            () -> Boolean.TRUE,
             new InternalTranslogFactory()
         );
 
@@ -146,6 +147,7 @@ public void testCreateCodecServiceFromFactory() {
             null,
             null,
             false,
+            () -> Boolean.TRUE,
             new InternalTranslogFactory()
         );
         assertNotNull(config.getCodec());