Only append tailed entries if needed. (grafana#2931)

* Only append tailed entries if needed.

This avoid to doing allocations on slices when there is no tailers.

We could potentially pooled the slice too but the tailing code is using the stream in an
async fashion so we can't put it back.

Signed-off-by: Cyril Tovena <[email protected]>

* cleanup

Signed-off-by: Cyril Tovena <[email protected]>

pkg/ingester/instance_test.go

@@ -240,7 +240,7 @@ func Test_SeriesQuery(t *testing.T) {
 }
 
 func entries(n int, t time.Time) []logproto.Entry {
-	var result []logproto.Entry
+	result := make([]logproto.Entry, 0, n)
 	for i := 0; i < n; i++ {
 		result = append(result, logproto.Entry{Timestamp: t, Line: fmt.Sprintf("hello %d", i)})
 		t = t.Add(time.Nanosecond)

pkg/ingester/stream.go

@@ -122,7 +122,15 @@ func (s *stream) Push(ctx context.Context, entries []logproto.Entry, synchronize
 		_, lastChunkTimestamp = s.chunks[len(s.chunks)-1].chunk.Bounds()
 	}
 
-	storedEntries := []logproto.Entry{}
+	s.tailerMtx.RLock()
+	hasTailers := len(s.tailers) != 0
+	s.tailerMtx.RUnlock()
+
+	var storedEntries []logproto.Entry
+	if hasTailers {
+		storedEntries = make([]logproto.Entry, 0, len(entries))
+	}
+
 	failedEntriesWithError := []entryWithError{}
 
 	// Don't fail on the first append error - if samples are sent out of order,
@@ -165,9 +173,12 @@ func (s *stream) Push(ctx context.Context, entries []logproto.Entry, synchronize
 			failedEntriesWithError = append(failedEntriesWithError, entryWithError{&entries[i], err})
 		} else {
 			// send only stored entries to tailers
-			storedEntries = append(storedEntries, entries[i])
+			if hasTailers {
+				storedEntries = append(storedEntries, entries[i])
+			}
 			lastChunkTimestamp = entries[i].Timestamp
-			s.lastLine = line{ts: lastChunkTimestamp, content: entries[i].Line}
+			s.lastLine.ts = lastChunkTimestamp
+			s.lastLine.content = entries[i].Line
 		}
 		chunk.lastUpdated = time.Now()
 	}

pkg/ingester/stream_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/weaveworks/common/httpgrpc"
 
 	"github.com/grafana/loki/pkg/chunkenc"
+	"github.com/grafana/loki/pkg/iter"
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/logql/log"
 )
@@ -140,3 +141,94 @@ func TestStreamIterator(t *testing.T) {
 	}
 
 }
+
+func Benchmark_PushStream(b *testing.B) {
+	ls := labels.Labels{
+		labels.Label{Name: "namespace", Value: "loki-dev"},
+		labels.Label{Name: "cluster", Value: "dev-us-central1"},
+		labels.Label{Name: "job", Value: "loki-dev/ingester"},
+		labels.Label{Name: "container", Value: "ingester"},
+	}
+	s := newStream(&Config{}, model.Fingerprint(0), ls, func() chunkenc.Chunk {
+		return &noopChunk{}
+	})
+	t, err := newTailer("foo", `{namespace="loki-dev"}`, &fakeTailServer{})
+	require.NoError(b, err)
+
+	go t.loop()
+	defer t.close()
+
+	s.tailers[1] = t
+	ctx := context.Background()
+	e := entries(100, time.Now())
+	b.ResetTimer()
+	b.ReportAllocs()
+
+	for n := 0; n < b.N; n++ {
+		require.NoError(b, s.Push(ctx, e, 0, 0))
+	}
+}
+
+type noopChunk struct {
+}
+
+func (c *noopChunk) Bounds() (time.Time, time.Time) {
+	return time.Time{}, time.Time{}
+}
+
+func (c *noopChunk) SpaceFor(_ *logproto.Entry) bool {
+	return true
+}
+
+func (c *noopChunk) Append(entry *logproto.Entry) error {
+	return nil
+}
+
+func (c *noopChunk) Size() int {
+	return 0
+}
+
+// UncompressedSize implements Chunk.
+func (c *noopChunk) UncompressedSize() int {
+	return c.Size()
+}
+
+// CompressedSize implements Chunk.
+func (c *noopChunk) CompressedSize() int {
+	return 0
+}
+
+// Utilization implements Chunk
+func (c *noopChunk) Utilization() float64 {
+	return 0
+}
+
+func (c *noopChunk) Encoding() chunkenc.Encoding { return chunkenc.EncNone }
+
+func (c *noopChunk) Iterator(_ context.Context, from, through time.Time, direction logproto.Direction, _ log.StreamPipeline) (iter.EntryIterator, error) {
+	return nil, nil
+}
+
+func (c *noopChunk) SampleIterator(_ context.Context, from, through time.Time, _ log.StreamSampleExtractor) iter.SampleIterator {
+	return nil
+}
+
+func (c *noopChunk) Bytes() ([]byte, error) {
+	return nil, nil
+}
+
+func (c *noopChunk) BytesWith(_ []byte) ([]byte, error) {
+	return nil, nil
+}
+
+func (c *noopChunk) Blocks(_ time.Time, _ time.Time) []chunkenc.Block {
+	return nil
+}
+
+func (c *noopChunk) BlockCount() int {
+	return 0
+}
+
+func (c *noopChunk) Close() error {
+	return nil
+}