Fix querying historic entries while live tailing logs

Querying logs in backward direction, then fetching all the logs upto the query limit and
then iterating over it in reverse order

pkg/iter/iterator.go

@@ -513,3 +513,54 @@ func ReadBatch(i EntryIterator, size uint32) (*logproto.QueryResponse, uint32, e
 	}
 	return &result, respSize, i.Error()
 }
+
+type entryWithLabels struct {
+	entry  logproto.Entry
+	labels string
+}
+
+type entryIteratorForward struct {
+	backwardIter      EntryIterator
+	cur               entryWithLabels
+	entriesWithLabels []entryWithLabels
+	loaded            bool
+	limit             uint32
+}
+
+// NewEntryIteratorBackward returns an iterator which loads all or upton N entries
+// of an existing iterator, and then iterates over them backward.
+func NewEntryIteratorForward(it EntryIterator, limit uint32) (EntryIterator, error) {
+	return &entryIteratorForward{entriesWithLabels: make([]entryWithLabels, 0, 1024), backwardIter: it, limit: limit}, it.Error()
+}
+
+func (i *entryIteratorForward) load() {
+	if !i.loaded {
+		i.loaded = true
+		for count := uint32(0); (i.limit == 0 || count < i.limit) && i.backwardIter.Next(); count++ {
+			i.entriesWithLabels = append(i.entriesWithLabels, entryWithLabels{i.backwardIter.Entry(), i.backwardIter.Labels()})
+		}
+		i.backwardIter.Close()
+	}
+}
+
+func (i *entryIteratorForward) Next() bool {
+	i.load()
+	if len(i.entriesWithLabels) == 0 {
+		i.entriesWithLabels = nil
+		return false
+	}
+	i.cur, i.entriesWithLabels = i.entriesWithLabels[len(i.entriesWithLabels)-1], i.entriesWithLabels[:len(i.entriesWithLabels)-1]
+	return true
+}
+
+func (i *entryIteratorForward) Entry() logproto.Entry {
+	return i.cur.entry
+}
+
+func (i *entryIteratorForward) Close() error { return nil }
+
+func (i *entryIteratorForward) Error() error { return nil }
+
+func (i *entryIteratorForward) Labels() string {
+	return i.cur.labels
+}

pkg/iter/iterator_test.go

@@ -252,3 +252,23 @@ func TestMostCommon(t *testing.T) {
 	}
 	require.Equal(t, "c", mostCommon(tuples).Entry.Line)
 }
+
+func TestEntryIteratorForward(t *testing.T) {
+	itr1 := mkStreamIterator(inverse(offset(testSize, identity)), defaultLabels)
+	itr2 := mkStreamIterator(inverse(offset(testSize, identity)), "{foobar: \"bazbar\"}")
+
+	heapIterator := NewHeapIterator([]EntryIterator{itr1, itr2}, logproto.BACKWARD)
+	forwardIterator, err := NewEntryIteratorForward(heapIterator, testSize)
+	require.NoError(t, err)
+
+	for i := int64((testSize / 2) + 1); i <= testSize; i++ {
+		assert.Equal(t, true, forwardIterator.Next())
+		assert.Equal(t, identity(i), forwardIterator.Entry(), fmt.Sprintln("iteration", i))
+		assert.Equal(t, true, forwardIterator.Next())
+		assert.Equal(t, identity(i), forwardIterator.Entry(), fmt.Sprintln("iteration", i))
+	}
+
+	assert.Equal(t, false, forwardIterator.Next())
+	assert.Equal(t, nil, forwardIterator.Error())
+	assert.NoError(t, forwardIterator.Close())
+}

pkg/querier/querier.go

@@ -270,7 +270,7 @@ func (q *Querier) Tail(ctx context.Context, req *logproto.TailRequest) (*Tailer,
 		Start:     req.Start,
 		End:       time.Now(),
 		Limit:     req.Limit,
-		Direction: logproto.FORWARD,
+		Direction: logproto.BACKWARD,
 		Regex:     req.Regex,
 	}
 	histIterators, err := q.getQueryIterators(queryCtx, &histReq)
@@ -278,10 +278,15 @@ func (q *Querier) Tail(ctx context.Context, req *logproto.TailRequest) (*Tailer,
 		return nil, err
 	}
 
+	reversedIterator, err := iter.NewEntryIteratorForward(iter.NewHeapIterator(histIterators, logproto.BACKWARD), req.Limit)
+	if err != nil {
+		return nil, err
+	}
+
 	return newTailer(
 		time.Duration(req.DelayFor)*time.Second,
 		tailClients,
-		histIterators,
+		reversedIterator,
 		func(from, to time.Time, labels string) (iterator iter.EntryIterator, e error) {
 			return q.queryDroppedStreams(queryCtx, req, from, to, labels)
 		},

pkg/querier/tail.go

@@ -327,13 +327,13 @@ func (t *Tailer) getCloseErrorChan() <-chan error {
 func newTailer(
 	delayFor time.Duration,
 	querierTailClients map[string]logproto.Querier_TailClient,
-	historicEntries []iter.EntryIterator,
+	historicEntries iter.EntryIterator,
 	queryDroppedStreams func(from, to time.Time, labels string) (iter.EntryIterator, error),
 	tailDisconnectedIngesters func([]string) (map[string]logproto.Querier_TailClient, error),
 	tailMaxDuration time.Duration,
 ) *Tailer {
 	t := Tailer{
-		openStreamIterator: iter.NewHeapIterator(historicEntries, logproto.FORWARD),
+		openStreamIterator: iter.NewHeapIterator([]iter.EntryIterator{}, logproto.FORWARD),
 		//droppedStreamsIterator:    &droppedStreamsIterator{},
 		querierTailClients:        querierTailClients,
 		queryDroppedStreams:       queryDroppedStreams,
@@ -344,6 +344,8 @@ func newTailer(
 		tailMaxDuration:           tailMaxDuration,
 	}
 
+	t.openStreamIterator.Push(historicEntries)
+
 	t.readTailClients()
 	go t.loop()
 	return &t