Use prometheus pool for line buffer.

pkg/chunkenc/gzip.go

@@ -476,8 +476,8 @@ type bufferedIterator struct {
 
 	err error
 
-	buf    *bytes.Buffer // The buffer for a single entry.
-	decBuf []byte        // The buffer for decoding the lengths.
+	buf    []byte // The buffer for a single entry.
+	decBuf []byte // The buffer for decoding the lengths.
 
 	closed bool
 
@@ -491,7 +491,6 @@ func newBufferedIterator(pool CompressionPool, b []byte, filter logql.Filter) *b
 		reader: r,
 		pool:   pool,
 		filter: filter,
-		buf:    BytesBufferPool.Get(),
 		decBuf: make([]byte, binary.MaxVarintLen64),
 	}
 }
@@ -529,25 +528,32 @@ func (si *bufferedIterator) moveNext() (int64, []byte, bool) {
 			return 0, nil, false
 		}
 	}
+	lineSize := int(l)
 
-	if si.buf.Cap() < int(l) {
-		si.buf.Grow(int(l) - si.buf.Cap())
+	// If the buffer is not yet initialize or too small, we get a new one.
+	if si.buf == nil || lineSize > cap(si.buf) {
+		// in case of replacement
+		if si.buf != nil {
+			BytesBufferPool.Put(si.buf)
+		}
+		si.buf = BytesBufferPool.Get(lineSize).([]byte)
 	}
 
-	n, err := si.s.Read(si.buf.Bytes()[:l])
+	// Then process reading the line.
+	n, err := si.s.Read(si.buf[:lineSize])
 	if err != nil && err != io.EOF {
 		si.err = err
 		return 0, nil, false
 	}
-	for n < int(l) {
-		r, err := si.s.Read(si.buf.Bytes()[n:l])
+	for n < lineSize {
+		r, err := si.s.Read(si.buf[n:lineSize])
 		if err != nil {
 			si.err = err
 			return 0, nil, false
 		}
 		n += r
 	}
-	return ts, si.buf.Bytes()[:l], true
+	return ts, si.buf[:lineSize], true
 }
 
 func (si *bufferedIterator) Entry() logproto.Entry {

pkg/chunkenc/pool.go

@@ -2,11 +2,11 @@ package chunkenc
 
 import (
 	"bufio"
-	"bytes"
 	"compress/gzip"
-
 	"io"
 	"sync"
+
+	"github.com/prometheus/prometheus/pkg/pool"
 )
 
 // CompressionPool is a pool of CompressionWriter and CompressionReader
@@ -28,7 +28,7 @@ var (
 		},
 	}
 	// BytesBufferPool is a bytes buffer used for lines decompressed.
-	BytesBufferPool = newBufferPoolWithSize(4096)
+	BytesBufferPool = pool.New(1024, 16384, 2, func(size int) interface{} { return make([]byte, 0, size) })
 )
 
 // GzipPool is a gun zip compression pool
@@ -92,23 +92,3 @@ func (bufPool *BufioReaderPool) Get(r io.Reader) *bufio.Reader {
 func (bufPool *BufioReaderPool) Put(b *bufio.Reader) {
 	bufPool.pool.Put(b)
 }
-
-type bufferPool struct {
-	pool sync.Pool
-}
-
-func newBufferPoolWithSize(size int) *bufferPool {
-	return &bufferPool{
-		pool: sync.Pool{
-			New: func() interface{} { return bytes.NewBuffer(make([]byte, size)) },
-		},
-	}
-}
-
-func (bp *bufferPool) Get() *bytes.Buffer {
-	return bp.pool.Get().(*bytes.Buffer)
-}
-
-func (bp *bufferPool) Put(b *bytes.Buffer) {
-	bp.pool.Put(b)
-}