Add timestamp for nrt points (#915)

Author: aprudhomme

clientlib/src/main/proto/yelp/nrtsearch/luceneserver.proto

@@ -1197,6 +1197,8 @@ message CopyState {
     repeated string completedMergeFiles = 7;
     // Primary generation
     int64 primaryGen = 8;
+    // Timestamp for this index version (epoch seconds), may be 0 for no timestamp
+    int64 timestampSec = 9;
 }
 
 // Metadata for multiple index files

grpc-gateway/luceneserver.pb.go

@@ -2821,6 +2821,11 @@
           "type": "string",
           "format": "int64",
           "title": "Primary generation"
+        },
+        "timestampSec": {
+          "type": "string",
+          "format": "int64",
+          "title": "Timestamp for this index version (epoch seconds), may be 0 for no timestamp"
         }
       },
       "title": "Holds incRef'd file level details for one point-in-time segment infos on the primary node"

grpc-gateway/luceneserver.swagger.json

@@ -26,6 +26,7 @@
 import com.yelp.nrtsearch.server.nrt.NRTPrimaryNode;
 import com.yelp.nrtsearch.server.state.GlobalState;
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Map;
 import org.apache.lucene.replicator.nrt.FileMetaData;
 import org.slf4j.Logger;
@@ -65,28 +66,29 @@ private CopyState handle(IndexState indexState, CopyStateRequest copyStateReques
       throw new RuntimeException("RecvCopyStateHandler invoked with Invalid Magic Number");
     }
     NRTPrimaryNode primaryNode = shardState.nrtPrimaryNode;
-    org.apache.lucene.replicator.nrt.CopyState copyState = null;
+    NRTPrimaryNode.CopyStateAndTimestamp copyStateAndTimestamp = null;
     try {
       // Caller does not have CopyState; we pull the latest NRT point:
-      copyState = primaryNode.getCopyState();
-      return RecvCopyStateHandler.writeCopyState(copyState);
+      copyStateAndTimestamp = primaryNode.getCopyStateAndTimestamp();
+      return RecvCopyStateHandler.writeCopyState(copyStateAndTimestamp);
     } catch (IOException e) {
       primaryNode.message("top: exception during fetch: " + e.getMessage());
       throw new RuntimeException(e);
     } finally {
-      if (copyState != null) {
+      if (copyStateAndTimestamp != null) {
         primaryNode.message("top: fetch: now release CopyState");
         try {
-          primaryNode.releaseCopyState(copyState);
+          primaryNode.releaseCopyState(copyStateAndTimestamp.copyState());
         } catch (IOException e) {
           throw new RuntimeException(e);
         }
       }
     }
   }
 
-  private static CopyState writeCopyState(org.apache.lucene.replicator.nrt.CopyState state)
-      throws IOException {
+  private static CopyState writeCopyState(
+      NRTPrimaryNode.CopyStateAndTimestamp copyStateAndTimestamp) throws IOException {
+    org.apache.lucene.replicator.nrt.CopyState state = copyStateAndTimestamp.copyState();
     CopyState.Builder builder = CopyState.newBuilder();
     builder.setInfoBytesLength(state.infosBytes().length);
     builder.setInfoBytes(ByteString.copyFrom(state.infosBytes(), 0, state.infosBytes().length));
@@ -105,6 +107,11 @@ private static CopyState writeCopyState(org.apache.lucene.replicator.nrt.CopySta
 
     builder.setPrimaryGen(state.primaryGen());
 
+    // If this index version has a timestamp, add it to the response
+    Instant timestamp = copyStateAndTimestamp.timestamp();
+    long timestampSeconds = timestamp != null ? timestamp.getEpochSecond() : 0;
+    builder.setTimestampSec(timestampSeconds);
+
     return builder.build();
   }
 

src/main/java/com/yelp/nrtsearch/server/handler/RecvCopyStateHandler.java

@@ -15,6 +15,7 @@
  */
 package com.yelp.nrtsearch.server.nrt;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.yelp.nrtsearch.server.grpc.FilesMetadata;
 import com.yelp.nrtsearch.server.grpc.ReplicationServerClient;
 import com.yelp.nrtsearch.server.grpc.TransferStatus;
@@ -27,6 +28,7 @@
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
 import java.io.PrintStream;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -64,6 +66,18 @@ public class NRTPrimaryNode extends PrimaryNode {
   final List<MergePreCopy> warmingSegments = Collections.synchronizedList(new ArrayList<>());
   final Queue<ReplicaDetails> replicasInfos = new ConcurrentLinkedQueue<>();
 
+  private long preRefreshVersion = -1;
+  private Instant refreshTimestamp;
+  private Instant indexVersionTimestamp;
+
+  /**
+   * Class to hold a copy state and its timestamp
+   *
+   * @param copyState lucene copy state
+   * @param timestamp timestamp
+   */
+  public record CopyStateAndTimestamp(CopyState copyState, Instant timestamp) {}
+
   public NRTPrimaryNode(
       IndexStateManager indexStateManager,
       HostPort hostPort,
@@ -80,6 +94,8 @@ public NRTPrimaryNode(
     this.indexName = indexStateManager.getCurrent().getName();
     this.indexStateManager = indexStateManager;
     this.nrtDataManager = nrtDataManager;
+    // initialize timestamp to that of index data loaded from the backend
+    setIndexVersionTimestamp(nrtDataManager.getLastPointTimestamp());
   }
 
   /**
@@ -584,4 +600,80 @@ public void close() throws IOException {
   public FileMetaData readLocalFileMetaData(String fileName) throws IOException {
     return NrtUtils.readOnceLocalFileMetaData(fileName, lastFileMetaData, this);
   }
+
+  @VisibleForTesting
+  synchronized void setPreRefreshVersion(long preRefreshVersion) {
+    this.preRefreshVersion = preRefreshVersion;
+  }
+
+  @VisibleForTesting
+  synchronized long getPreRefreshVersion() {
+    return preRefreshVersion;
+  }
+
+  @VisibleForTesting
+  synchronized void setRefreshTimestamp(Instant refreshTimestamp) {
+    this.refreshTimestamp = refreshTimestamp;
+  }
+
+  @VisibleForTesting
+  synchronized Instant getRefreshTimestamp() {
+    return refreshTimestamp;
+  }
+
+  @VisibleForTesting
+  synchronized void setIndexVersionTimestamp(Instant indexVersionTimestamp) {
+    this.indexVersionTimestamp = indexVersionTimestamp;
+  }
+
+  @VisibleForTesting
+  synchronized Instant getIndexVersionTimestamp() {
+    return indexVersionTimestamp;
+  }
+
+  /**
+   * Function that wraps the call to the super method and updates the index version timestamp.
+   *
+   * @return true if the index version changed
+   * @throws IOException
+   */
+  @Override
+  public boolean flushAndRefresh() throws IOException {
+    synchronized (this) {
+      // get the loaded index version before the refresh
+      setPreRefreshVersion(getCopyStateVersion());
+      // create a new timestamp to use for any index version created by this refresh
+      setRefreshTimestamp(Instant.now());
+    }
+
+    boolean refreshed = super.flushAndRefresh();
+
+    synchronized (this) {
+      // if the index version was updated, also update the main timestamp
+      if (refreshed) {
+        setIndexVersionTimestamp(getRefreshTimestamp());
+      }
+      // reset version to indicate no refresh in progress
+      setPreRefreshVersion(-1);
+    }
+    return refreshed;
+  }
+
+  /**
+   * Get the copy state representing the currently loaded index version, and its timestamp.
+   *
+   * @return copy state and timestamp
+   * @throws IOException
+   */
+  public synchronized CopyStateAndTimestamp getCopyStateAndTimestamp() throws IOException {
+    CopyState copyState = getCopyState();
+    Instant timestamp;
+    if (getPreRefreshVersion() != -1 && copyState.version() > getPreRefreshVersion()) {
+      // the index version has changed, but the timestamp is not updated yet, use the refresh value
+      timestamp = getRefreshTimestamp();
+    } else {
+      timestamp = getIndexVersionTimestamp();
+    }
+    return new CopyStateAndTimestamp(copyState, timestamp);
+  }
 }

src/main/java/com/yelp/nrtsearch/server/nrt/NRTPrimaryNode.java

@@ -24,9 +24,11 @@
 import com.yelp.nrtsearch.server.nrt.jobs.CopyJobManager;
 import com.yelp.nrtsearch.server.nrt.jobs.GrpcCopyJobManager;
 import com.yelp.nrtsearch.server.nrt.jobs.RemoteCopyJobManager;
+import com.yelp.nrtsearch.server.nrt.jobs.SimpleCopyJob;
 import com.yelp.nrtsearch.server.utils.HostPort;
 import java.io.IOException;
 import java.io.PrintStream;
+import java.time.Instant;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.lucene.index.SegmentInfos;
@@ -221,6 +223,14 @@ protected void finishNRTCopy(CopyJob job, long startNS) throws IOException {
           .observe((System.nanoTime() - startNS) / 1000000.0);
       NrtMetrics.nrtPointSize.labelValues(indexName).observe(job.getTotalBytesCopied());
       NrtMetrics.searcherVersion.labelValues(indexName).set(job.getCopyState().version());
+
+      // if the job is a simple copy job, read out the index data timestamp and update the metric
+      if (job instanceof SimpleCopyJob simpleCopyJob) {
+        Instant timestamp = simpleCopyJob.getTimestamp();
+        if (timestamp != null) {
+          NrtMetrics.indexTimestampSec.labelValues(indexName).set(timestamp.getEpochSecond());
+        }
+      }
     }
   }
 

src/main/java/com/yelp/nrtsearch/server/nrt/NRTReplicaNode.java

@@ -129,8 +129,8 @@ NrtPointState getLastPointState() {
     return lastPointState;
   }
 
-  @VisibleForTesting
-  Instant getLastPointTimestamp() {
+  /** Get the timestamp associated with the last loaded index version */
+  public Instant getLastPointTimestamp() {
     return lastPointTimestamp;
   }
 

src/main/java/com/yelp/nrtsearch/server/nrt/NrtDataManager.java

@@ -18,9 +18,11 @@
 import static com.yelp.nrtsearch.server.nrt.NrtUtils.readFilesMetaData;
 
 import com.yelp.nrtsearch.server.grpc.ReplicationServerClient;
+import com.yelp.nrtsearch.server.nrt.NRTPrimaryNode;
 import com.yelp.nrtsearch.server.nrt.NRTReplicaNode;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.time.Instant;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -72,26 +74,26 @@ public CopyJob newCopyJob(
           "Cannot create new copy job, primary connection not available");
     }
 
-    CopyState copyState;
+    NRTPrimaryNode.CopyStateAndTimestamp copyStateAndTimestamp;
 
     // sendMeFiles(?) (we dont need this, just send Index,replica, and request for copy State)
     if (files == null) {
       // No incoming CopyState: ask primary for latest one now
       try {
         // Exceptions in here mean something went wrong talking over the socket, which are fine
         // (e.g. primary node crashed):
-        copyState = getCopyStateFromPrimary();
+        copyStateAndTimestamp = getCopyStateFromPrimary();
       } catch (Throwable t) {
         throw new NodeCommunicationException("exc while reading files to copy", t);
       }
-      files = copyState.files();
+      files = copyStateAndTimestamp.copyState().files();
     } else {
-      copyState = null;
+      copyStateAndTimestamp = null;
     }
     return new SimpleCopyJob(
         reason,
         primaryAddress,
-        copyState,
+        copyStateAndTimestamp,
         replicaNode,
         files,
         highPriority,
@@ -104,15 +106,15 @@ public CopyJob newCopyJob(
   @Override
   public void finishNRTCopy(CopyJob copyJob) throws IOException {}
 
-  private CopyState getCopyStateFromPrimary() throws IOException {
+  private NRTPrimaryNode.CopyStateAndTimestamp getCopyStateFromPrimary() throws IOException {
     com.yelp.nrtsearch.server.grpc.CopyState copyState =
         primaryAddress.recvCopyState(indexName, indexId, replicaId);
     return readCopyState(copyState);
   }
 
   /** Pulls CopyState off the wire */
-  private static CopyState readCopyState(com.yelp.nrtsearch.server.grpc.CopyState copyState)
-      throws IOException {
+  private static NRTPrimaryNode.CopyStateAndTimestamp readCopyState(
+      com.yelp.nrtsearch.server.grpc.CopyState copyState) throws IOException {
 
     // Decode a new CopyState
     byte[] infosBytes = new byte[copyState.getInfoBytesLength()];
@@ -125,7 +127,11 @@ private static CopyState readCopyState(com.yelp.nrtsearch.server.grpc.CopyState
     Set<String> completedMergeFiles = new HashSet<>(copyState.getCompletedMergeFilesList());
     long primaryGen = copyState.getPrimaryGen();
 
-    return new CopyState(files, version, gen, infosBytes, completedMergeFiles, primaryGen, null);
+    CopyState luceneCopyState =
+        new CopyState(files, version, gen, infosBytes, completedMergeFiles, primaryGen, null);
+    Instant timestamp =
+        copyState.getTimestampSec() > 0 ? Instant.ofEpochMilli(copyState.getTimestampSec()) : null;
+    return new NRTPrimaryNode.CopyStateAndTimestamp(luceneCopyState, timestamp);
   }
 
   @Override

src/main/java/com/yelp/nrtsearch/server/nrt/jobs/GrpcCopyJobManager.java

@@ -19,8 +19,10 @@
 import com.yelp.nrtsearch.server.grpc.RawFileChunk;
 import com.yelp.nrtsearch.server.grpc.ReplicationServerClient;
 import com.yelp.nrtsearch.server.monitoring.NrtMetrics;
+import com.yelp.nrtsearch.server.nrt.NRTPrimaryNode;
 import io.grpc.stub.StreamObserver;
 import java.io.IOException;
+import java.time.Instant;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Locale;
@@ -42,7 +44,7 @@
 public class SimpleCopyJob extends VisitableCopyJob {
   private static final Logger logger = LoggerFactory.getLogger(SimpleCopyJob.class);
 
-  private final CopyState copyState;
+  private final NRTPrimaryNode.CopyStateAndTimestamp copyStateAndTimestamp;
   private final ReplicationServerClient primaryAddres;
   private final String indexName;
   private final String indexId;
@@ -52,7 +54,7 @@ public class SimpleCopyJob extends VisitableCopyJob {
   public SimpleCopyJob(
       String reason,
       ReplicationServerClient primaryAddress,
-      CopyState copyState,
+      NRTPrimaryNode.CopyStateAndTimestamp copyStateAndTimestamp,
       ReplicaNode dest,
       Map<String, FileMetaData> files,
       boolean highPriority,
@@ -62,7 +64,7 @@ public SimpleCopyJob(
       boolean ackedCopy)
       throws IOException {
     super(reason, files, dest, highPriority, onceDone);
-    this.copyState = copyState;
+    this.copyStateAndTimestamp = copyStateAndTimestamp;
     this.primaryAddres = primaryAddress;
     this.indexName = indexName;
     this.indexId = indexId;
@@ -189,7 +191,12 @@ public Set<String> getFileNames() {
 
   @Override
   public CopyState getCopyState() {
-    return copyState;
+    return copyStateAndTimestamp != null ? copyStateAndTimestamp.copyState() : null;
+  }
+
+  /** Get the index version timestamp, or null if no timestamp is available. */
+  public Instant getTimestamp() {
+    return copyStateAndTimestamp != null ? copyStateAndTimestamp.timestamp() : null;
   }
 
   @Override