Fix bug in replication lag computation.

This change fixes a bug with replication lag computation to correctly use epoch reference point with Instant.now() and DateUtils.
This change also fixes pruning logic to correctly remove the latest synced to checkpoint from tracking.  Previously we would only
prune up to the latest.  This ensures that when a new checkpoint is eventually received we aren't incorrectly computing lag from the synced-to checkpoint.

Signed-off-by: Marc Handalian <[email protected]>

Author: mch2

server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java

@@ -14,6 +14,7 @@
 import org.apache.lucene.index.CorruptIndexException;
 import org.opensearch.OpenSearchCorruptionException;
 import org.opensearch.common.SetOnce;
+import org.opensearch.common.time.DateUtils;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.concurrent.AbstractRunnable;
 import org.opensearch.common.util.concurrent.ConcurrentCollections;
@@ -30,11 +31,12 @@
 import org.opensearch.threadpool.ThreadPool;
 
 import java.io.IOException;
+import java.time.Duration;
+import java.time.Instant;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.TimeUnit;
 
 /**
  * This class is responsible for managing segment replication events on replicas.
@@ -49,7 +51,7 @@ public class SegmentReplicator {
 
     private final ReplicationCollection<SegmentReplicationTarget> onGoingReplications;
     private final Map<ShardId, SegmentReplicationState> completedReplications = ConcurrentCollections.newConcurrentMap();
-    private final ConcurrentMap<ShardId, ConcurrentNavigableMap<Long, ReplicationCheckpointStats>> replicationCheckpointStats =
+    protected final ConcurrentMap<ShardId, ConcurrentNavigableMap<Long, ReplicationCheckpointStats>> replicationCheckpointStats =
         ConcurrentCollections.newConcurrentMap();
     private final ConcurrentMap<ShardId, ReplicationCheckpoint> primaryCheckpoint = ConcurrentCollections.newConcurrentMap();
 
@@ -130,9 +132,8 @@ public ReplicationStats getSegmentReplicationStats(final ShardId shardId) {
 
         long bytesBehind = highestEntry.getValue().getBytesBehind();
         long replicationLag = bytesBehind > 0L
-            ? TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lowestEntry.getValue().getTimestamp())
+            ? Duration.ofNanos(DateUtils.toLong(Instant.now()) - lowestEntry.getValue().getTimestamp()).toMillis()
             : 0;
-
         return new ReplicationStats(bytesBehind, bytesBehind, replicationLag);
     }
 
@@ -180,7 +181,7 @@ protected void pruneCheckpointsUpToLastSync(final IndexShard indexShard) {
             );
 
             if (existingCheckpointStats != null && !existingCheckpointStats.isEmpty()) {
-                existingCheckpointStats.keySet().removeIf(key -> key < segmentInfoVersion);
+                existingCheckpointStats.keySet().removeIf(key -> key <= segmentInfoVersion);
                 Map.Entry<Long, ReplicationCheckpointStats> lastEntry = existingCheckpointStats.lastEntry();
                 if (lastEntry != null) {
                     lastEntry.getValue().setBytesBehind(calculateBytesBehind(latestCheckpoint, indexReplicationCheckPoint));

server/src/main/java/org/opensearch/indices/replication/checkpoint/ReplicationCheckpoint.java

@@ -11,6 +11,7 @@
 import org.opensearch.Version;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.PublicApi;
+import org.opensearch.common.time.DateUtils;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.common.io.stream.Writeable;
@@ -19,6 +20,7 @@
 import org.opensearch.index.store.StoreFileMetadata;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
@@ -56,11 +58,11 @@ private ReplicationCheckpoint(ShardId shardId, String codec) {
         length = 0L;
         this.codec = codec;
         this.metadataMap = Collections.emptyMap();
-        this.createdTimeStamp = System.nanoTime();
+        this.createdTimeStamp = DateUtils.toLong(Instant.now());
     }
 
     public ReplicationCheckpoint(ShardId shardId, long primaryTerm, long segmentsGen, long segmentInfosVersion, String codec) {
-        this(shardId, primaryTerm, segmentsGen, segmentInfosVersion, 0L, codec, Collections.emptyMap(), System.nanoTime());
+        this(shardId, primaryTerm, segmentsGen, segmentInfosVersion, 0L, codec, Collections.emptyMap(), DateUtils.toLong(Instant.now()));
     }
 
     public ReplicationCheckpoint(
@@ -79,7 +81,7 @@ public ReplicationCheckpoint(
         this.length = length;
         this.codec = codec;
         this.metadataMap = metadataMap;
-        this.createdTimeStamp = System.nanoTime();
+        this.createdTimeStamp = DateUtils.toLong(Instant.now());
     }
 
     public ReplicationCheckpoint(

server/src/test/java/org/opensearch/indices/replication/SegmentReplicatorTests.java

@@ -21,6 +21,7 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.lucene.Lucene;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.time.DateUtils;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.index.ReplicationStats;
@@ -38,6 +39,7 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -232,7 +234,7 @@ public void testGetSegmentReplicationStats_WhileOnGoingReplicationAndPrimaryRefr
             1000,
             "",
             stringStoreFileMetadataMapOne,
-            System.nanoTime() - TimeUnit.MINUTES.toNanos(1)
+            DateUtils.toLong(Instant.now())
         );
 
         IndexShard replicaShard = mock(IndexShard.class);
@@ -260,7 +262,7 @@ public void testGetSegmentReplicationStats_WhileOnGoingReplicationAndPrimaryRefr
             200,
             "",
             stringStoreFileMetadataMapTwo,
-            System.nanoTime() - TimeUnit.MINUTES.toNanos(1)
+            DateUtils.toLong(Instant.now())
         );
 
         segmentReplicator.updateReplicationCheckpointStats(thirdReplicationCheckpoint, replicaShard);
@@ -276,6 +278,16 @@ public void testGetSegmentReplicationStats_WhileOnGoingReplicationAndPrimaryRefr
         assertEquals(200, replicationStatsSecond.totalBytesBehind);
         assertEquals(200, replicationStatsSecond.maxBytesBehind);
         assertTrue(replicationStatsSecond.maxReplicationLag > 0);
+
+        // shard finished syncing to last checkpoint (sis 3)
+        when(replicaShard.getLatestReplicationCheckpoint()).thenReturn(thirdReplicationCheckpoint);
+        segmentReplicator.pruneCheckpointsUpToLastSync(replicaShard);
+        ReplicationStats finalStats = segmentReplicator.getSegmentReplicationStats(shardId);
+        assertEquals(0, finalStats.totalBytesBehind);
+        assertEquals(0, finalStats.maxBytesBehind);
+        assertEquals(0, finalStats.maxReplicationLag);
+        // shard is up to date, should not have any tracked stats
+        assertTrue(segmentReplicator.replicationCheckpointStats.get(shardId).isEmpty());
     }
 
     public void testGetSegmentReplicationStats_WhenCheckPointReceivedOutOfOrder() {