Skip segrep lag computation for shard copies on docrep nodes

Signed-off-by: Shourya Dutta Biswas <[email protected]>
opensearch-project · Jun 10, 2024 · 819ffae · 819ffae
1 parent 1b36ee4
commit 819ffae
Show file tree

Hide file tree

Showing 2 changed files with 49 additions and 2 deletions.
diff --git a/.../src/internalClusterTest/java/org/opensearch/remotemigration/RemoteDualReplicationIT.java b/.../src/internalClusterTest/java/org/opensearch/remotemigration/RemoteDualReplicationIT.java
@@ -18,6 +18,7 @@
 import org.opensearch.cluster.routing.allocation.command.MoveAllocationCommand;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.IndexService;
+import org.opensearch.index.ReplicationStats;
 import org.opensearch.index.remote.RemoteSegmentStats;
 import org.opensearch.index.seqno.RetentionLease;
 import org.opensearch.index.seqno.RetentionLeases;
@@ -665,6 +666,43 @@ public void testFailoverRemotePrimaryToDocrepReplicaReseedToRemotePrimary() thro
         });
     }
 
+    /*
+    Performs the same experiment as testRemotePrimaryDocRepReplica.
+
+    This ensures that the primary shard for the index has moved over to remote
+    enabled node whereas the replica copy is still left behind on the docrep nodes
+
+    At this stage, segrep lag computation shouldn't consider the docrep shard copy while calculating bytes lag
+     */
+    public void testZeroSegrepLagForShardsWithMixedReplicationGroup() throws Exception {
+        testRemotePrimaryDocRepReplica();
+        String remoteNodeName = internalCluster().client()
+            .admin()
+            .cluster()
+            .prepareNodesStats()
+            .get()
+            .getNodes()
+            .stream()
+            .filter(nodeStats -> nodeStats.getNode().isRemoteStoreNode())
+            .findFirst()
+            .get()
+            .getNode()
+            .getName();
+        ReplicationStats replicationStats = internalCluster().client()
+            .admin()
+            .cluster()
+            .prepareNodesStats(remoteNodeName)
+            .get()
+            .getNodes()
+            .get(0)
+            .getIndices()
+            .getSegments()
+            .getReplicationStats();
+        assertEquals(0, replicationStats.getMaxBytesBehind());
+        assertEquals(0, replicationStats.getTotalBytesBehind());
+        assertEquals(0, replicationStats.getMaxReplicationLag());
+    }
+
     private void assertReplicaAndPrimaryConsistency(String indexName, int firstBatch, int secondBatch) throws Exception {
         assertBusy(() -> {
             Map<ShardRouting, ShardStats> shardStatsMap = internalCluster().client()

diff --git a/server/src/main/java/org/opensearch/index/seqno/ReplicationTracker.java b/server/src/main/java/org/opensearch/index/seqno/ReplicationTracker.java
@@ -1319,14 +1319,23 @@ && isPrimaryRelocation(e.getKey()) == false
     public synchronized Set<SegmentReplicationShardStats> getSegmentReplicationStats() {
         assert indexSettings.isSegRepEnabledOrRemoteNode();
         if (primaryMode) {
+            // Check if the current primary shard is migrating to remote and
+            // all the other shard copies of the same index still hasn't completely moved over
+            // to the remote enabled nodes
+            boolean migratingPrimaryOnRemoteNode = indexSettings.isAssignedOnRemoteNode() && indexSettings.isRemoteStoreEnabled() == false;
             return this.checkpoints.entrySet()
                 .stream()
-                // filter out this shard's allocation id, any shards that are out of sync or unavailable (shard marked in-sync but has not
-                // been assigned to a node).
+                /* Filter out:
+                - This shard's allocation id
+                - Any shards that are out of sync or unavailable (shard marked in-sync but has not been assigned to a node).
+                - (For remote store enabled clusters) Any shard that is not yet migrated to remote store enabled nodes during migration
+                 */
                 .filter(
                     entry -> entry.getKey().equals(this.shardAllocationId) == false
                         && entry.getValue().inSync
                         && replicationGroup.getUnavailableInSyncShards().contains(entry.getKey()) == false
+                        && (migratingPrimaryOnRemoteNode
+                            && isShardOnRemoteEnabledNode.apply(routingTable.getByAllocationId(entry.getKey()).currentNodeId()))
                         && isPrimaryRelocation(entry.getKey()) == false
                 )
                 .map(entry -> buildShardStats(entry.getKey(), entry.getValue()))