Full cluster replication POC

Author: ankitkala

server/src/main/java/org/opensearch/action/ActionModule.java

@@ -439,9 +439,18 @@
 import org.opensearch.rest.action.search.RestMultiSearchAction;
 import org.opensearch.rest.action.search.RestSearchAction;
 import org.opensearch.rest.action.search.RestSearchScrollAction;
+import org.opensearch.xreplication.actions.index.StartCCRIndexTaskAction;
+import org.opensearch.xreplication.actions.index.TransportStartCCRIndexTaskAction;
+import org.opensearch.xreplication.actions.start.RestStartCCRAction;
+import org.opensearch.xreplication.actions.start.StartCCRAction;
+import org.opensearch.xreplication.actions.start.TransportStartCCRAction;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.usage.UsageService;
+import org.opensearch.xreplication.actions.followers.StartCCRFollowerTaskAction;
+import org.opensearch.xreplication.actions.followers.TransportStartCCRFollowerTaskAction;
+import org.opensearch.xreplication.actions.syncsegments.SyncFromLeaderAction;
+import org.opensearch.xreplication.actions.syncsegments.TransportSyncFromLeaderAction;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -731,6 +740,14 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(DecommissionAction.INSTANCE, TransportDecommissionAction.class);
         actions.register(GetDecommissionStateAction.INSTANCE, TransportGetDecommissionStateAction.class);
         actions.register(DeleteDecommissionStateAction.INSTANCE, TransportDeleteDecommissionStateAction.class);
+
+
+        // Tansport actions for Full Cluster Replication
+        actions.register(StartCCRAction.INSTANCE, TransportStartCCRAction.class);
+        actions.register(StartCCRFollowerTaskAction.INSTANCE, TransportStartCCRFollowerTaskAction.class);
+        actions.register(StartCCRIndexTaskAction.INSTANCE, TransportStartCCRIndexTaskAction.class);
+        actions.register(SyncFromLeaderAction.INSTANCE, TransportSyncFromLeaderAction.class);
+
         return unmodifiableMap(actions.getRegistry());
     }
 
@@ -915,6 +932,9 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestPitSegmentsAction(nodesInCluster));
         registerHandler.accept(new RestDeleteDecommissionStateAction());
 
+        // xcluster
+        registerHandler.accept(new RestStartCCRAction());
+
         for (ActionPlugin plugin : actionPlugins) {
             for (RestHandler handler : plugin.getRestHandlers(
                 settings,

server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

@@ -495,6 +495,16 @@ public Iterator<Setting<?>> settings() {
         Property.IndexScope,
         Property.Final
     );
+    public static final String CCR_REPLICATING_FROM_INDEX = "index.replication.fcr.replicating_from";
+    public static final Setting<String> CCR_REPLICATING_FROM_INDEX_SETTING = new Setting<>(CCR_REPLICATING_FROM_INDEX, "", Function.identity(), Property.IndexScope);
+
+    public static final String CCR_REMOTE_PATH = "index.replication.fcr.remote_path";
+    public static final Setting<String> CCR_REMOTE_PATH_SETTING = new Setting<>(
+        CCR_REMOTE_PATH,
+        "",
+        Function.identity(),
+        Property.IndexScope
+    );
 
     public static final String SETTING_AUTO_EXPAND_REPLICAS = "index.auto_expand_replicas";
     public static final Setting<AutoExpandReplicas> INDEX_AUTO_EXPAND_REPLICAS_SETTING = AutoExpandReplicas.SETTING;

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

@@ -196,6 +196,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexSettings.INDEX_MERGE_ON_FLUSH_ENABLED,
                 IndexSettings.INDEX_MERGE_ON_FLUSH_MAX_FULL_FLUSH_MERGE_WAIT_TIME,
                 IndexSettings.INDEX_MERGE_ON_FLUSH_POLICY,
+                IndexMetadata.CCR_REPLICATING_FROM_INDEX_SETTING,
 
                 // validate that built-in similarities don't get redefined
                 Setting.groupSetting("index.similarity.", (s) -> {

server/src/main/java/org/opensearch/common/util/FeatureFlags.java

@@ -27,6 +27,12 @@ public class FeatureFlags {
      */
     public static final String REPLICATION_TYPE = "opensearch.experimental.feature.replication_type.enabled";
 
+    /**
+     * Gates the functionality of full cluster replication.
+     * Once the feature is ready for production release, this feature flag can be removed.
+     */
+    public static final String X_REPLICATION = "opensearch.experimental.feature.full_cluster_replication.enabled";
+
     /**
      * Gates the visibility of the index setting that allows persisting data to remote store along with local disk.
      * Once the feature is ready for production release, this feature flag can be removed.
@@ -75,6 +81,7 @@ public static void initializeFeatureFlags(Settings openSearchSettings) {
      * and false otherwise.
      */
     public static boolean isEnabled(String featureFlagName) {
+        if (REPLICATION_TYPE.equals(featureFlagName) || REMOTE_STORE.equals(featureFlagName) || X_REPLICATION.equals(featureFlagName)) return true;
         if ("true".equalsIgnoreCase(System.getProperty(featureFlagName))) {
             // TODO: Remove the if condition once FeatureFlags are only supported via opensearch.yml
             return true;

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

@@ -62,6 +62,7 @@
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
 
+import static org.opensearch.cluster.metadata.IndexMetadata.CCR_REMOTE_PATH_SETTING;
 import static org.opensearch.common.util.FeatureFlags.SEARCHABLE_SNAPSHOT_EXTENDED_COMPATIBILITY;
 import static org.opensearch.index.mapper.MapperService.INDEX_MAPPING_DEPTH_LIMIT_SETTING;
 import static org.opensearch.index.mapper.MapperService.INDEX_MAPPING_FIELD_NAME_LENGTH_LIMIT_SETTING;
@@ -587,6 +588,8 @@ public final class IndexSettings {
     private final int numberOfShards;
     private final ReplicationType replicationType;
     private final boolean isRemoteStoreEnabled;
+    private final boolean isCCRReplicatingIndex;
+    private final String ccrReplicatingFrom;
     private final boolean isRemoteTranslogStoreEnabled;
     private final TimeValue remoteTranslogUploadBufferInterval;
     private final String remoteStoreTranslogRepository;
@@ -761,6 +764,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         isRemoteStoreEnabled = settings.getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false);
         isRemoteTranslogStoreEnabled = settings.getAsBoolean(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_ENABLED, false);
         remoteStoreTranslogRepository = settings.get(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY);
+        ccrReplicatingFrom = settings.get(IndexMetadata.CCR_REPLICATING_FROM_INDEX, "");
+        isCCRReplicatingIndex = !Strings.isNullOrEmpty(ccrReplicatingFrom);
         remoteTranslogUploadBufferInterval = settings.getAsTime(
             IndexMetadata.SETTING_REMOTE_TRANSLOG_BUFFER_INTERVAL,
             TimeValue.timeValueMillis(100)
@@ -1024,6 +1029,13 @@ public boolean isRemoteStoreEnabled() {
         return isRemoteStoreEnabled;
     }
 
+    public boolean isCCRReplicatingIndex() {
+        return isCCRReplicatingIndex;
+    }
+
+    public String getCCRReplicatingFrom() {
+        return ccrReplicatingFrom;
+    }
     /**
      * Returns remote store repository configured for this index.
      */
@@ -1565,4 +1577,8 @@ private void setMergeOnFlushPolicy(String policy) {
     public Optional<UnaryOperator<MergePolicy>> getMergeOnFlushPolicy() {
         return Optional.ofNullable(mergeOnFlushPolicy);
     }
+
+    public String getCCRRemotePath() {
+        return CCR_REMOTE_PATH_SETTING.get(settings);
+    }
 }

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

@@ -408,6 +408,11 @@ public boolean isReadOnlyReplica() {
         return indexSettings.isSegRepEnabled() && isReadOnlyReplica;
     }
 
+
+    public boolean isReadOnlyPrimary() {
+        return indexSettings.isSegRepEnabled() && indexSettings.isRemoteStoreEnabled() && indexSettings.isCCRReplicatingIndex() && !isReadOnlyReplica;
+    }
+
     /**
      * Returns the underlying primaryModeSupplier.
      * @return the primary mode supplier.

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

@@ -256,6 +256,7 @@ public InternalTranslogManager translogManager() {
             throttle = new IndexThrottle();
             try {
                 store.trimUnsafeCommits(engineConfig.getTranslogConfig().getTranslogPath());
+                //TODO: fix this. Reading translogUUID from segments won't work if the segments are from a different cluster(CCR).
                 final Map<String, String> userData = store.readLastCommittedSegmentsInfo().getUserData();
                 final String translogUUID = Objects.requireNonNull(userData.get(Translog.TRANSLOG_UUID_KEY));
                 TranslogEventListener internalTranslogEventListener = new TranslogEventListener() {

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

@@ -17,7 +17,8 @@
 public class NRTReplicationEngineFactory implements EngineFactory {
     @Override
     public Engine newReadWriteEngine(EngineConfig config) {
-        if (config.isReadOnlyReplica()) {
+        // Load NRTReplicationEngine for primaries on CCR Follower as well.
+        if (config.isReadOnlyReplica() || config.isReadOnlyPrimary()) {
             return new NRTReplicationEngine(config);
         }
         return new InternalEngine(config);

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

@@ -162,6 +162,7 @@
 import org.opensearch.index.store.StoreStats;
 import org.opensearch.index.translog.Translog;
 import org.opensearch.index.translog.TranslogConfig;
+import org.opensearch.index.translog.TranslogCorruptedException;
 import org.opensearch.index.translog.TranslogFactory;
 import org.opensearch.index.translog.TranslogStats;
 import org.opensearch.index.warmer.ShardIndexWarmerService;
@@ -2185,7 +2186,7 @@ private void innerOpenEngineAndTranslog(LongSupplier globalCheckpointSupplier) t
             assert currentEngineReference.get() == null : "engine is running";
             verifyNotClosed();
             if (indexSettings.isRemoteStoreEnabled()) {
-                syncSegmentsFromRemoteSegmentStore(false, true, false);
+                syncSegmentsFromRemoteSegmentStore(false, true, true);
             }
             // we must create a new engine under mutex (see IndexShard#snapshotStoreMetadata).
             final Engine newEngine = engineFactory.newReadWriteEngine(config);
@@ -3502,10 +3503,12 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
 
         final List<ReferenceManager.RefreshListener> internalRefreshListener = new ArrayList<>();
         internalRefreshListener.add(new RefreshMetricUpdater(refreshMetric));
-        if (isRemoteStoreEnabled()) {
+        // Skip SegRep refresh listener for CCR follower indices.
+        if (isRemoteStoreEnabled() && !indexSettings.isCCRReplicatingIndex()) {
             internalRefreshListener.add(new RemoteStoreRefreshListener(this, remoteSegmentNotificationPublisher));
         }
-        if (this.checkpointPublisher != null && indexSettings.isSegRepEnabled() && shardRouting.primary() && !indexSettings.isRemoteStoreEnabled()) {
+        // Skip SegRep refresh listener for CCR leader as well as follower indices.
+        if (this.checkpointPublisher != null && indexSettings.isSegRepEnabled() && shardRouting.primary() && !indexSettings.isRemoteStoreEnabled() && !indexSettings.isCCRReplicatingIndex()) {
             internalRefreshListener.add(new CheckpointRefreshListener(this, this.checkpointPublisher));
         }
 
@@ -4352,7 +4355,7 @@ public void close() throws IOException {
             };
             IOUtils.close(currentEngineReference.getAndSet(readOnlyEngine));
             if (indexSettings.isRemoteStoreEnabled()) {
-                syncSegmentsFromRemoteSegmentStore(false, true, false);
+                syncSegmentsFromRemoteSegmentStore(false, true, true);
             }
             newEngineReference.set(engineFactory.newReadWriteEngine(newEngineConfig(replicationTracker)));
             onNewEngine(newEngineReference.get());
@@ -4456,13 +4459,15 @@ public void syncSegmentsFromRemoteSegmentStore(boolean overrideLocal, boolean re
                         indexInput,
                         Long.parseLong(segmentInfosSnapshotFilename.split("__")[1])
                     );
+
                     long processedLocalCheckpoint = Long.parseLong(infosSnapshot.getUserData().get(LOCAL_CHECKPOINT_KEY));
                     if (shouldCommit) {
-                        finalizeReplication(infosSnapshot);
-                        store.cleanupAndPreserveLatestCommitPoint("finalize - clean with in memory infos", infosSnapshot);
+                        store.commitSegmentInfos(infosSnapshot, processedLocalCheckpoint, processedLocalCheckpoint);
                     }
                     else {
-                        store.commitSegmentInfos(infosSnapshot, processedLocalCheckpoint, processedLocalCheckpoint);
+                        // We don't need to trigger a commit for segment copy from primaries(like SegRep with remote store, CCR)
+                        finalizeReplication(infosSnapshot);
+                        store.cleanupAndPreserveLatestCommitPoint("finalize - clean with in memory infos", infosSnapshot);
                     }
                 }
             }

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

@@ -11,7 +11,7 @@
 import org.opensearch.common.inject.Inject;
 import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint;
 import org.opensearch.indices.replication.checkpoint.SegmentReplicationCheckpointPublisher;
-
+import org.opensearch.xreplication.actions.notifysecondary.NotifyCCRFollowersAction;
 
 /**
  * Hook to publish notification after primary uploads segments to the remote store.
@@ -20,18 +20,22 @@
  */
 public class RemoteStoreSegmentUploadNotificationPublisher {
     private final SegmentReplicationCheckpointPublisher segRepPublisher;
+    private final NotifyCCRFollowersAction xReplicatePublisher;
 
     @Inject
-    public RemoteStoreSegmentUploadNotificationPublisher(SegmentReplicationCheckpointPublisher segRepPublisher) {
+    public RemoteStoreSegmentUploadNotificationPublisher(SegmentReplicationCheckpointPublisher segRepPublisher, NotifyCCRFollowersAction xReplicatePublisher) {
         this.segRepPublisher = segRepPublisher;
+        this.xReplicatePublisher = xReplicatePublisher;
     }
 
+    // Notify replicas and CCR followers after the segments have been uploaded to the remote store by primary(during refresh).
     public void notifySegmentUpload(IndexShard indexShard, ReplicationCheckpoint checkpoint) {
-        // TODO: Add separate publisher for CCR.
         // we don't call indexShard.getLatestReplicationCheckpoint() as it might have a newer refreshed checkpoint.
         // Instead we send the one which has been uploaded to remote store.
+        // TODO: Parallise both the notifications.
         if (segRepPublisher != null) segRepPublisher.publish(indexShard, checkpoint);
+        if (xReplicatePublisher != null) xReplicatePublisher.publish(indexShard, checkpoint);
     }
 
-    public static final RemoteStoreSegmentUploadNotificationPublisher EMPTY = new RemoteStoreSegmentUploadNotificationPublisher(null);
+    public static final RemoteStoreSegmentUploadNotificationPublisher EMPTY = new RemoteStoreSegmentUploadNotificationPublisher(null, null);
 }

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

@@ -457,7 +457,7 @@ private void recoverFromRemoteStore(IndexShard indexShard, Repository repository
         remoteStore.incRef();
         try {
             // Download segments from remote segment store
-            indexShard.syncSegmentsFromRemoteSegmentStore(true, true, false);
+            indexShard.syncSegmentsFromRemoteSegmentStore(true, true, true);
 
             if (store.directory().listAll().length == 0) {
                 store.createEmpty(indexShard.indexSettings().getIndexVersionCreated().luceneVersion);

server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectoryFactory.java

@@ -41,7 +41,12 @@ public Directory newDirectory(IndexSettings indexSettings, ShardPath path) throw
         try (Repository repository = repositoriesService.get().repository(repositoryName)) {
             assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository";
             BlobPath commonBlobPath = ((BlobStoreRepository) repository).basePath();
-            commonBlobPath = commonBlobPath.add(indexSettings.getIndex().getUUID())
+            String indexUUID = indexSettings.getIndex().getUUID();
+            // override the follower's remote store path to leader index's uuid.
+            if (indexSettings.isCCRReplicatingIndex()) {
+                indexUUID = indexSettings.getCCRReplicatingFrom();
+            }
+            commonBlobPath = commonBlobPath.add(indexUUID)
                 .add(String.valueOf(path.getShardId().getId()))
                 .add("segments");
 

server/src/main/java/org/opensearch/index/store/remote/directory/RemoteSnapshotDirectoryFactory.java

@@ -73,10 +73,7 @@ private Future<RemoteSnapshotDirectory> createRemoteSnapshotDirectoryFromSnapsho
         ShardPath localShardPath,
         BlobStoreRepository blobStoreRepository
     ) throws IOException {
-        final BlobPath blobPath = blobStoreRepository.basePath()
-            .add("indices")
-            .add(IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID.get(indexSettings.getSettings()))
-            .add(Integer.toString(localShardPath.getShardId().getId()));
+        final BlobPath blobPath = getBlobPath(indexSettings, localShardPath, blobStoreRepository);
         final SnapshotId snapshotId = new SnapshotId(
             IndexSettings.SEARCHABLE_SNAPSHOT_ID_NAME.get(indexSettings.getSettings()),
             IndexSettings.SEARCHABLE_SNAPSHOT_ID_UUID.get(indexSettings.getSettings())
@@ -94,4 +91,15 @@ private Future<RemoteSnapshotDirectory> createRemoteSnapshotDirectoryFromSnapsho
             return new RemoteSnapshotDirectory(snapshot, localStoreDir, transferManager);
         });
     }
+
+    private BlobPath getBlobPath(IndexSettings indexSettings, ShardPath localShardPath, BlobStoreRepository blobStoreRepository) {
+        if(indexSettings.isCCRReplicatingIndex()) {
+            return blobStoreRepository.basePath().add(indexSettings.getCCRRemotePath());
+        } else {
+            return blobStoreRepository.basePath()
+                .add("indices")
+                .add(IndexSettings.SEARCHABLE_SNAPSHOT_INDEX_ID.get(indexSettings.getSettings()))
+                .add(Integer.toString(localShardPath.getShardId().getId()));
+        }
+    }
 }

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

@@ -32,6 +32,7 @@
 
 package org.opensearch.index.translog;
 
+import org.apache.logging.log4j.Logger;
 import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.index.IndexFormatTooNewException;
@@ -43,6 +44,7 @@
 import org.opensearch.common.io.stream.InputStreamStreamInput;
 import org.opensearch.common.io.stream.OutputStreamStreamOutput;
 import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.logging.Loggers;
 
 import java.io.EOFException;
 import java.io.IOException;
@@ -180,13 +182,16 @@ static TranslogHeader read(final String translogUUID, final Path path, final Fil
 
             // verify UUID only after checksum, to ensure that UUID is not corrupted
             final BytesRef expectedUUID = new BytesRef(translogUUID);
+            Logger logger = Loggers.getLogger(TranslogHeader.class, "");
+            // Skipping the check for now.
+            /*
             if (uuid.bytesEquals(expectedUUID) == false) {
                 throw new TranslogCorruptedException(
                     path.toString(),
                     "expected shard UUID " + expectedUUID + " but got: " + uuid + " this translog file belongs to a different translog"
                 );
             }
-
+            */
             return new TranslogHeader(translogUUID, primaryTerm, headerSizeInBytes);
         } catch (EOFException e) {
             throw new TranslogCorruptedException(path.toString(), "translog header truncated", e);