prevents duplicate log lines from being replayed. closes #3378

pkg/ingester/checkpoint.go

@@ -269,6 +269,9 @@ func (s *streamIterator) Next() bool {
 	s.current.Fingerprint = uint64(stream.fp)
 	s.current.Labels = client.FromLabelsToLabelAdapters(stream.labels)
 
+	s.current.To = stream.lastLine.ts
+	s.current.LastLine = stream.lastLine.content
+
 	return true
 }
 

pkg/ingester/checkpoint.proto

@@ -30,4 +30,7 @@ message Series {
   uint64 fingerprint = 2;
   repeated cortex.LabelPair labels = 3 [(gogoproto.nullable) = false, (gogoproto.customtype) = "github.com/cortexproject/cortex/pkg/ingester/client.LabelAdapter"];
   repeated Chunk chunks = 4 [(gogoproto.nullable) = false];
+  // Last timestamp of the last chunk.
+  google.protobuf.Timestamp to = 5 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
+  string lastLine = 6;
 }

pkg/ingester/encoding_test.go

@@ -243,6 +243,8 @@ func Test_EncodingCheckpoint(t *testing.T) {
 		UserID:      "fake",
 		Fingerprint: 123,
 		Labels:      client.FromLabelsToLabelAdapters(ls),
+		To:          time.Unix(10, 0),
+		LastLine:    "lastLine",
 		Chunks: []Chunk{
 			{
 				From:        from,
@@ -275,12 +277,17 @@ func Test_EncodingCheckpoint(t *testing.T) {
 	outChunks := out.Chunks
 	out.Chunks = nil
 
+	zero := time.Unix(0, 0)
+
+	require.Equal(t, true, s.To.Equal(out.To))
+	s.To = zero
+	out.To = zero
+
 	require.Equal(t, s, out)
 	require.Equal(t, len(sChunks), len(outChunks))
 	for i, exp := range sChunks {
 
 		got := outChunks[i]
-		zero := time.Unix(0, 0)
 		// Issues diffing zero-value time.Locations against nil ones.
 		// Check/override them individually so that other fields get tested in an extensible manner.
 		require.Equal(t, true, exp.From.Equal(got.From))

pkg/ingester/recovery.go

@@ -123,6 +123,8 @@ func (r *ingesterRecoverer) Series(series *Series) error {
 		}
 
 		bytesAdded, entriesAdded, err := stream.setChunks(series.Chunks)
+		stream.lastLine.ts = series.To
+		stream.lastLine.content = series.LastLine
 
 		if err != nil {
 			return err