Adds WAL support (experimental)

pkg/chunkenc/memchunk.go

@@ -152,17 +152,17 @@ func (hb *headBlock) serialise(pool WriterPool) ([]byte, error) {
 	return outBuf.Bytes(), nil
 }
 
-// CheckpointBytes is used for WAL checkpointing
+// CheckpointBytes serializes a headblock to []byte. This is used by the WAL checkpointing,
+// which does not want to mutate a chunk by cutting it (otherwise risking content address changes), but
+// needs to serialize/deserialize the data to disk to ensure data durability.
 func (hb *headBlock) CheckpointBytes(version byte) ([]byte, error) {
-	// wB is eventually returned via buf.Bytes(), don't return it to the pool.
-	wB := BytesBufferPool.Get(1 << 10).([]byte)
 	encB := BytesBufferPool.Get(1 << 10).([]byte)
 
 	defer func() {
 		BytesBufferPool.Put(encB[:0])
 	}()
 
-	buf := bytes.NewBuffer(wB[:0])
+	buf := bytes.NewBuffer(make([]byte, 0, 1<<10))
 	eb := encbuf{b: encB}
 
 	eb.putByte(version)

pkg/ingester/checkpoint.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/cortexproject/cortex/pkg/ingester/client"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
 	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
@@ -95,13 +96,14 @@ func decodeCheckpointRecord(rec []byte, s *Series) error {
 	}
 }
 
-func encodeWithTypeHeader(m proto.Message, typ RecordType, b []byte) ([]byte, error) {
+func encodeWithTypeHeader(m proto.Message, typ RecordType) ([]byte, error) {
 	buf, err := proto.Marshal(m)
 	if err != nil {
-		return b, err
+		return nil, err
 	}
 
-	b = append(b[:0], byte(typ))
+	b := make([]byte, 0, len(buf)+1)
+	b = append(b, byte(typ))
 	b = append(b, buf...)
 	return b, nil
 }
@@ -112,7 +114,7 @@ type SeriesWithErr struct {
 }
 
 type SeriesIter interface {
-	Num() int
+	Count() int
 	Iter() <-chan *SeriesWithErr
 	Stop()
 }
@@ -130,7 +132,7 @@ func newIngesterSeriesIter(ing *Ingester) *ingesterSeriesIter {
 	}
 }
 
-func (i *ingesterSeriesIter) Num() (ct int) {
+func (i *ingesterSeriesIter) Count() (ct int) {
 	for _, inst := range i.ing.getInstances() {
 		ct += inst.numStreams()
 	}
@@ -234,8 +236,8 @@ func (w *WALCheckpointWriter) Advance() (bool, error) {
 	if err := os.MkdirAll(checkpointDirTemp, 0777); err != nil {
 		return false, errors.Wrap(err, "create checkpoint dir")
 	}
-	// checkpoint, err ;= wal.NewSize()
-	checkpoint, err := wal.NewSize(nil, nil, checkpointDirTemp, walSegmentSize, false)
+
+	checkpoint, err := wal.NewSize(log.With(util.Logger, "component", "checkpoint_wal"), nil, checkpointDirTemp, walSegmentSize, false)
 	if err != nil {
 		return false, errors.Wrap(err, "open checkpoint")
 	}
@@ -248,7 +250,7 @@ func (w *WALCheckpointWriter) Advance() (bool, error) {
 }
 
 func (w *WALCheckpointWriter) Write(s *Series) error {
-	b, err := encodeWithTypeHeader(s, CheckpointRecord, recordPool.GetBytes()[:0])
+	b, err := encodeWithTypeHeader(s, CheckpointRecord)
 	if err != nil {
 		return err
 	}
@@ -430,7 +432,7 @@ func (c *Checkpointer) PerformCheckpoint() (err error) {
 	}()
 	// signal whether checkpoint writes should be amortized or burst
 	var immediate bool
-	n := c.iter.Num()
+	n := c.iter.Count()
 	if n < 1 {
 		return c.writer.Close(false)
 	}

pkg/ingester/encoding.go

@@ -133,15 +133,11 @@ func decodeEntries(b []byte, rec *WALRecord) error {
 
 	for len(dec.B) > 0 && dec.Err() == nil {
 		refEntries := RefEntries{
-			Ref:     dec.Be64(),
-			Entries: recordPool.GetEntries(),
+			Ref: dec.Be64(),
 		}
 
 		nEntries := dec.Uvarint()
 		rem := nEntries
-		if cap(refEntries.Entries) < nEntries {
-			refEntries.Entries = make([]logproto.Entry, 0, nEntries)
-		}
 		for ; dec.Err() == nil && rem > 0; rem-- {
 			timeOffset := dec.Varint64()
 			lineLength := dec.Uvarint()
@@ -181,8 +177,6 @@ func decodeWALRecord(b []byte, walRec *WALRecord) (err error) {
 		t      = RecordType(decbuf.Byte())
 	)
 
-	walRec.Reset()
-
 	switch t {
 	case WALRecordSeries:
 		userID = decbuf.UvarintStr()

pkg/ingester/encoding_test.go

@@ -38,11 +38,11 @@ func Test_Encoding_Series(t *testing.T) {
 
 	buf := record.encodeSeries(nil)
 
-	var decoded WALRecord
+	decoded := recordPool.GetRecord()
 
-	err := decodeWALRecord(buf, &decoded)
+	err := decodeWALRecord(buf, decoded)
 	require.Nil(t, err)
-	require.Equal(t, record, &decoded)
+	require.Equal(t, record, decoded)
 }
 
 func Test_Encoding_Entries(t *testing.T) {
@@ -81,25 +81,23 @@ func Test_Encoding_Entries(t *testing.T) {
 
 	buf := record.encodeEntries(nil)
 
-	var decoded WALRecord
+	decoded := recordPool.GetRecord()
 
-	err := decodeWALRecord(buf, &decoded)
+	err := decodeWALRecord(buf, decoded)
 	require.Nil(t, err)
-	require.Equal(t, record, &decoded)
+	require.Equal(t, record, decoded)
 }
 
-func fillChunk(c chunkenc.Chunk) int64 {
+func fillChunk(t *testing.T, c chunkenc.Chunk) int64 {
+	t.Helper()
 	var i, inserted int64
 	entry := &logproto.Entry{
 		Timestamp: time.Unix(0, 0),
 		Line:      "entry for line 0",
 	}
 
 	for c.SpaceFor(entry) {
-		err := c.Append(entry)
-		if err != nil {
-			panic(err)
-		}
+		require.NoError(t, c.Append(entry))
 		i++
 		inserted += int64(len(entry.Line))
 		entry.Timestamp = time.Unix(0, i)
@@ -120,7 +118,7 @@ func Test_EncodingChunks(t *testing.T) {
 
 	conf := dummyConf()
 	c := chunkenc.NewMemChunk(chunkenc.EncGZIP, conf.BlockSize, conf.TargetChunkSize)
-	fillChunk(c)
+	fillChunk(t, c)
 
 	from := []chunkDesc{
 		{
@@ -187,7 +185,7 @@ func Test_EncodingCheckpoint(t *testing.T) {
 		},
 	}
 
-	b, err := encodeWithTypeHeader(s, CheckpointRecord, recordPool.GetBytes()[:0])
+	b, err := encodeWithTypeHeader(s, CheckpointRecord)
 	require.Nil(t, err)
 
 	out := &Series{}

pkg/ingester/ingester.go

@@ -31,6 +31,7 @@ import (
 	"github.com/grafana/loki/pkg/logql/stats"
 	"github.com/grafana/loki/pkg/storage"
 	"github.com/grafana/loki/pkg/storage/stores/shipper"
+	errUtil "github.com/grafana/loki/pkg/util"
 	listutil "github.com/grafana/loki/pkg/util"
 	"github.com/grafana/loki/pkg/util/validation"
 )
@@ -147,7 +148,6 @@ type Ingester struct {
 
 	metrics *ingesterMetrics
 
-	// WAL
 	wal WAL
 }
 
@@ -297,9 +297,9 @@ func (i *Ingester) running(ctx context.Context) error {
 // At this point, loop no longer runs, but flushers are still running.
 func (i *Ingester) stopping(_ error) error {
 	i.stopIncomingRequests()
-	i.wal.Stop()
-
-	err := services.StopAndAwaitTerminated(context.Background(), i.lifecycler)
+	var errs errUtil.MultiError
+	errs.Add(i.wal.Stop())
+	errs.Add(services.StopAndAwaitTerminated(context.Background(), i.lifecycler))
 
 	// Normally, flushers are stopped via lifecycler (in transferOut), but if lifecycler fails,
 	// we better stop them.
@@ -308,7 +308,7 @@ func (i *Ingester) stopping(_ error) error {
 	}
 	i.flushQueuesDone.Wait()
 
-	return err
+	return errs.Err()
 }
 
 func (i *Ingester) loop() {

pkg/ingester/recovery.go

@@ -353,8 +353,8 @@ func recoverGeneric(
 		inputs = append(inputs, make(chan recoveryInput))
 
 		go func(input <-chan recoveryInput) {
+			defer wg.Done()
 			process(recoverer, input, errCh)
-			wg.Done()
 		}(inputs[i])
 
 	}

pkg/ingester/stream.go

@@ -143,7 +143,7 @@ func (s *stream) Push(
 		_, lastChunkTimestamp = s.chunks[len(s.chunks)-1].chunk.Bounds()
 	}
 
-	storedEntries := make([]logproto.Entry, 0, len(entries))
+	storedEntries := recordPool.GetEntries()
 	failedEntriesWithError := []entryWithError{}
 
 	// Don't fail on the first append error - if samples are sent out of order,

pkg/ingester/tailer.go

@@ -111,6 +111,7 @@ func (t *tailer) loop() {
 }
 
 func (t *tailer) send(stream logproto.Stream, lbs labels.Labels) {
+	defer recordPool.PutEntries(stream.Entries)
 	if t.isClosed() {
 		return
 	}

pkg/ingester/wal.go

@@ -50,13 +50,13 @@ type WAL interface {
 	// Log marshalls the records and writes it into the WAL.
 	Log(*WALRecord) error
 	// Stop stops all the WAL operations.
-	Stop()
+	Stop() error
 }
 
 type noopWAL struct{}
 
 func (noopWAL) Log(*WALRecord) error { return nil }
-func (noopWAL) Stop()                {}
+func (noopWAL) Stop() error          { return nil }
 
 type walWrapper struct {
 	cfg        WALConfig
@@ -127,11 +127,12 @@ func (w *walWrapper) Log(record *WALRecord) error {
 	}
 }
 
-func (w *walWrapper) Stop() {
+func (w *walWrapper) Stop() error {
 	close(w.quit)
 	w.wait.Wait()
-	_ = w.wal.Close()
+	err := w.wal.Close()
 	level.Info(util.Logger).Log("msg", "stopped", "component", "wal")
+	return err
 }
 
 func (w *walWrapper) checkpointWriter() *WALCheckpointWriter {