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.StartIndexTaskAction;
+import org.opensearch.xreplication.actions.index.TransportStartIndexTaskAction;
+import org.opensearch.xreplication.actions.start.RestXReplicateAction;
+import org.opensearch.xreplication.actions.start.StartXReplication;
+import org.opensearch.xreplication.actions.start.TransportXReplicateAction;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.usage.UsageService;
+import org.opensearch.xreplication.actions.followers.StartFollowersAction;
+import org.opensearch.xreplication.actions.followers.TransportStartFollowerAction;
+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);
+
+
+        // Full Cluster Replication
+        actions.register(StartXReplication.INSTANCE, TransportXReplicateAction.class);
+        actions.register(StartFollowersAction.INSTANCE, TransportStartFollowerAction.class);
+        actions.register(StartIndexTaskAction.INSTANCE, TransportStartIndexTaskAction.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 RestXReplicateAction());
+
         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,12 @@ public boolean isReadOnlyReplica() {
         return indexSettings.isSegRepEnabled() && isReadOnlyReplica;
     }
 
+
+    public boolean isReadOnlyPrimary() {
+        //TODO: verify on segrep enabled.
+        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,7 @@
 public class NRTReplicationEngineFactory implements EngineFactory {
     @Override
     public Engine newReadWriteEngine(EngineConfig config) {
-        if (config.isReadOnlyReplica()) {
+        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,10 @@ private EngineConfig newEngineConfig(LongSupplier globalCheckpointSupplier) thro
 
         final List<ReferenceManager.RefreshListener> internalRefreshListener = new ArrayList<>();
         internalRefreshListener.add(new RefreshMetricUpdater(refreshMetric));
-        if (isRemoteStoreEnabled()) {
+        if (isRemoteStoreEnabled() && !indexSettings.isCCRReplicatingIndex()) {
             internalRefreshListener.add(new RemoteStoreRefreshListener(this, remoteSegmentNotificationPublisher));
         }
-        if (this.checkpointPublisher != null && indexSettings.isSegRepEnabled() && shardRouting.primary() && !indexSettings.isRemoteStoreEnabled()) {
+        if (this.checkpointPublisher != null && indexSettings.isSegRepEnabled() && shardRouting.primary() && !indexSettings.isRemoteStoreEnabled() && !indexSettings.isCCRReplicatingIndex()) {
             internalRefreshListener.add(new CheckpointRefreshListener(this, this.checkpointPublisher));
         }
 
@@ -4352,7 +4353,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 +4457,14 @@ 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);
+                        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.NotifySecondariesAction;
 
 /**
  * Hook to publish notification after primary uploads segments to the remote store.
@@ -20,18 +20,21 @@
  */
 public class RemoteStoreSegmentUploadNotificationPublisher {
     private final SegmentReplicationCheckpointPublisher segRepPublisher;
+    private final NotifySecondariesAction xReplicatePublisher;
 
     @Inject
-    public RemoteStoreSegmentUploadNotificationPublisher(SegmentReplicationCheckpointPublisher segRepPublisher) {
+    public RemoteStoreSegmentUploadNotificationPublisher(SegmentReplicationCheckpointPublisher segRepPublisher, NotifySecondariesAction xReplicatePublisher) {
         this.segRepPublisher = segRepPublisher;
+        this.xReplicatePublisher = xReplicatePublisher;
     }
 
     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);
 }