live-tailing: preload all the historic entries before query context i…

…s cancelled (#862)

* preload all the historic entries before query context is cancelled

* fix lint error

pkg/iter/iterator.go

@@ -529,8 +529,18 @@ type entryIteratorForward struct {
 
 // 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()
+// preload entries when they are being queried with a timeout
+func NewEntryIteratorForward(it EntryIterator, limit uint32, preload bool) (EntryIterator, error) {
+	itr, err := &entryIteratorForward{entriesWithLabels: make([]entryWithLabels, 0, 1024), backwardIter: it, limit: limit}, it.Error()
+	if err != nil {
+		return nil, err
+	}
+
+	if preload {
+		itr.load()
+	}
+
+	return itr, nil
 }
 
 func (i *entryIteratorForward) load() {

pkg/iter/iterator_test.go

@@ -258,7 +258,7 @@ func TestEntryIteratorForward(t *testing.T) {
 	itr2 := mkStreamIterator(inverse(offset(testSize, identity)), "{foobar: \"bazbar\"}")
 
 	heapIterator := NewHeapIterator([]EntryIterator{itr1, itr2}, logproto.BACKWARD)
-	forwardIterator, err := NewEntryIteratorForward(heapIterator, testSize)
+	forwardIterator, err := NewEntryIteratorForward(heapIterator, testSize, false)
 	require.NoError(t, err)
 
 	for i := int64((testSize / 2) + 1); i <= testSize; i++ {

pkg/querier/querier.go

@@ -295,7 +295,7 @@ 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)
+	reversedIterator, err := iter.NewEntryIteratorForward(iter.NewHeapIterator(histIterators, logproto.BACKWARD), req.Limit, true)
 	if err != nil {
 		return nil, err
 	}