Only append tailed entries if needed.

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
+}