Repository: prometheus/tsdb
Pull request #528 (2019-02-18 11:03:19)

Test to corrupt segments mid-WAL, repair and check we can read the correct number of records.
The test fails against master, as the underlying segmentReader has already moved on to the next segment.

Also included - the fix.  Make the segmentReader pad short segments with zeros, and only advance to next segment on the call to `Read` _after_ completely consuming this one; previously it would increment `curr` on the call to `Read` which failed, causing the returns `CorruptionErr` to reference the wrong segment.

Signed-off-by: Tom Wilkie <tom.wilkie@gmail.com>

tomwilkie: I think this is an improvement, but I'm trying to think if there are still corrupt WALs which could fail repair - for instance, a WAL where a short segment causes the `io.ReadFull` to consume into the next segment.  Given the io.ReadFull is bounded by a page, the only case I can think of are segments containing records spanning pages - which shouldn't happen, but does.

To make this more robust (defense in depth), I think we should:
- Change repair to not depend on the the index of the WAL segment from the error, but rebuild the entire WAL (which would be slow)
- Get rid of repair all together, and when we find a corrupt WAL delete the entire thing and write out a new checkpoint (after all, we've just replayed the WAL into the head).
- Change the WAL code to not append to the existing segment, but always start a new segment.

WDYT?
tomwilkie: Thanks @krasi-georgiev!
cstyan: > a WAL where a short segment causes the `io.ReadFull` to consume into the next segment. Given the io.ReadFull is bounded by a page, the only case I can think of are segments containing records spanning pages - which shouldn't happen, but does.

I don't think this is actually true. I suspect the corruptions we were seeing (both torn pages and failed crc checks) were because we were reading data at incorrect offsets in the file because the previous corrupt segment hadn't been padded with 0's.
tomwilkie: > I don't think this is actually true. I suspect the corruptions we were seeing (both torn pages and failed crc checks) were because we were reading data at incorrect offsets in the file because the previous corrupt segment hadn't been padded with 0's.

We definitely have a segment with records spanning pages (and not just from concatenating segments, which isn't something the LiveReader does).  But I have no idea how we got into this state, and its kinda unrelated.
tomwilkie: Thanks for the feedback @cstyan @krasi-georgiev !  I think I've covered it all, left me know if I've missed anything.
krasi-georgiev: > To make this more robust (defense in depth), I think we should:

> * Change repair to not depend on the the index of the WAL segment from the error, but rebuild the entire WAL (which would be slow)

> * Get rid of repair all together, and when we find a corrupt WAL delete the entire thing and write out a new checkpoint (after all, we've just replayed the WAL into the head).

this one sounds as the best solution, but maybe do it in a separate PR if it happens that the current implementation causes other problems.

> * Change the WAL code to not append to the existing segment, but always start a new segment.


tomwilkie: > this one sounds as the best solution, but maybe do it in a separate PR if it happens that the current implementation causes other problems.

For sure!  I was thinking aloud there.  I'll follow up with a ticket for this. 

Patch:
diff --git a/wal/wal.go b/wal/wal.go
index 1ab6334..c778256 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -681,20 +681,20 @@ func NewSegmentsRangeReader(sr ...SegmentRange) (io.ReadCloser, error) {
 
 // segmentBufReader is a buffered reader that reads in multiples of pages.
 // The main purpose is that we are able to track segment and offset for
-// corruption reporting.
+// corruption reporting.  We have to be careful not to increment curr too
+// early, as it is used by Reader.Err() to tell Repair which segment is corrupt.
+// As such we pad the end of non-page align segments with zeros.
 type segmentBufReader struct {
 	buf  *bufio.Reader
 	segs []*Segment
-	cur  int
-	off  int
-	more bool
+	cur  int // Index into segs.
+	off  int // Offset of read data into current segment.
 }
 
 func newSegmentBufReader(segs ...*Segment) *segmentBufReader {
 	return &segmentBufReader{
-		buf:  bufio.NewReaderSize(nil, 16*pageSize),
+		buf:  bufio.NewReaderSize(segs[0], 16*pageSize),
 		segs: segs,
-		cur:  -1,
 	}
 }
 
@@ -707,25 +707,39 @@ func (r *segmentBufReader) Close() (err error) {
 	return err
 }
 
+// Read implements io.Reader.
 func (r *segmentBufReader) Read(b []byte) (n int, err error) {
-	if !r.more {
-		if r.cur+1 >= len(r.segs) {
-			return 0, io.EOF
-		}
-		r.cur++
-		r.off = 0
-		r.more = true
-		r.buf.Reset(r.segs[r.cur])
-	}
 	n, err = r.buf.Read(b)
 	r.off += n
-	if err != io.EOF {
+
+	// If we succeeded, or hit a non-EOF, we can stop.
+	if err == nil || err != io.EOF {
 		return n, err
 	}
-	// Just return what we read so far, but don't signal EOF.
-	// Only unset more so we don't invalidate the current segment and
-	// offset before the next read.
-	r.more = false
+
+	// We hit EOF; fake out zero padding at the end of short segments, so we
+	// don't increment curr too early and report the wrong segment as corrupt.
+	if r.off%pageSize != 0 {
+		i := 0
+		for ; n+i < len(b) && (r.off+i)%pageSize != 0; i++ {
+			b[n+i] = 0
+		}
+
+		// Return early, even if we didn't fill b.
+		r.off += i
+		return n + i, nil
+	}
+
+	// There is no more deta left in the curr segment and there are no more
+	// segments left.  Return EOF.
+	if r.cur+1 >= len(r.segs) {
+		return n, io.EOF
+	}
+
+	// Move to next segment.
+	r.cur++
+	r.off = 0
+	r.buf.Reset(r.segs[r.cur])
 	return n, nil
 }
 


Test patch:
diff --git a/testutil/logging.go b/testutil/logging.go
new file mode 100644
index 0000000..839b866
--- /dev/null
+++ b/testutil/logging.go
@@ -0,0 +1,35 @@
+// Copyright 2019 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package testutil
+
+import (
+	"testing"
+
+	"github.com/go-kit/kit/log"
+)
+
+type logger struct {
+	t *testing.T
+}
+
+// NewLogger returns a gokit compatible Logger which calls t.Log.
+func NewLogger(t *testing.T) log.Logger {
+	return logger{t: t}
+}
+
+// Log implements log.Logger.
+func (t logger) Log(keyvals ...interface{}) error {
+	t.t.Log(keyvals...)
+	return nil
+}
diff --git a/wal/wal_test.go b/wal/wal_test.go
index 737c44c..c767bb5 100644
--- a/wal/wal_test.go
+++ b/wal/wal_test.go
@@ -24,6 +24,7 @@ import (
 	"math/rand"
 	"os"
 	"path"
+	"path/filepath"
 	"sync"
 	"testing"
 	"time"
@@ -225,7 +226,7 @@ func TestReader_Live(t *testing.T) {
 }
 
 func TestWAL_FuzzWriteRead_Live(t *testing.T) {
-	const count = 5000
+	const count = 500
 	var input [][]byte
 	lock := sync.RWMutex{}
 	var recs [][]byte
@@ -547,6 +548,129 @@ func TestWAL_Repair(t *testing.T) {
 	}
 }
 
+// TestCorruptAndCarryOn writes a multi-segment WAL; corrupts the first segment and
+// ensures that an error during reading that segment are correctly repaired before
+// moving to write more records to the WAL.
+func TestCorruptAndCarryOn(t *testing.T) {
+	dir, err := ioutil.TempDir("", "wal_repair")
+	testutil.Ok(t, err)
+	defer os.RemoveAll(dir)
+
+	var (
+		logger      = testutil.NewLogger(t)
+		segmentSize = pageSize * 3
+		recordSize  = (pageSize / 3) - recordHeaderSize
+	)
+
+	// Produce a WAL with a two segments of 3 pages with 3 records each,
+	// so when we truncate the file we're guaranteed to split a record.
+	{
+		w, err := NewSize(logger, nil, dir, segmentSize)
+		testutil.Ok(t, err)
+
+		for i := 0; i < 18; i++ {
+			buf := make([]byte, recordSize)
+			_, err := rand.Read(buf)
+			testutil.Ok(t, err)
+
+			err = w.Log(buf)
+			testutil.Ok(t, err)
+		}
+
+		err = w.Close()
+		testutil.Ok(t, err)
+	}
+
+	// Check all the segments are the correct size.
+	{
+		segments, err := listSegments(dir)
+		testutil.Ok(t, err)
+		for _, segment := range segments {
+			f, err := os.OpenFile(filepath.Join(dir, fmt.Sprintf("%08d", segment.index)), os.O_RDONLY, 0666)
+			testutil.Ok(t, err)
+
+			fi, err := f.Stat()
+			testutil.Ok(t, err)
+
+			t.Log("segment", segment.index, "size", fi.Size())
+			testutil.Equals(t, int64(segmentSize), fi.Size())
+
+			err = f.Close()
+			testutil.Ok(t, err)
+		}
+	}
+
+	// Truncate the first file, splitting the middle record in the second
+	// page in half, leaving 4 valid records.
+	{
+		f, err := os.OpenFile(filepath.Join(dir, fmt.Sprintf("%08d", 0)), os.O_RDWR, 0666)
+		testutil.Ok(t, err)
+
+		fi, err := f.Stat()
+		testutil.Ok(t, err)
+		testutil.Equals(t, int64(segmentSize), fi.Size())
+
+		err = f.Truncate(int64(segmentSize / 2))
+		testutil.Ok(t, err)
+
+		err = f.Close()
+		testutil.Ok(t, err)
+	}
+
+	// Now try and repair this WAL, and write 5 more records to it.
+	{
+		sr, err := NewSegmentsReader(dir)
+		testutil.Ok(t, err)
+
+		reader := NewReader(sr)
+		i := 0
+		for ; i < 4 && reader.Next(); i++ {
+			testutil.Equals(t, recordSize, len(reader.Record()))
+		}
+		testutil.Equals(t, 4, i, "not enough records")
+		testutil.Assert(t, !reader.Next(), "unexpected record")
+
+		corruptionErr := reader.Err()
+		testutil.Assert(t, corruptionErr != nil, "expected error")
+
+		err = sr.Close()
+		testutil.Ok(t, err)
+
+		w, err := NewSize(logger, nil, dir, segmentSize)
+		testutil.Ok(t, err)
+
+		err = w.Repair(corruptionErr)
+		testutil.Ok(t, err)
+
+		for i := 0; i < 5; i++ {
+			buf := make([]byte, recordSize)
+			_, err := rand.Read(buf)
+			testutil.Ok(t, err)
+
+			err = w.Log(buf)
+			testutil.Ok(t, err)
+		}
+
+		err = w.Close()
+		testutil.Ok(t, err)
+	}
+
+	// Replay the WAL. Should get 9 records.
+	{
+		sr, err := NewSegmentsReader(dir)
+		testutil.Ok(t, err)
+
+		reader := NewReader(sr)
+		i := 0
+		for ; i < 9 && reader.Next(); i++ {
+			testutil.Equals(t, recordSize, len(reader.Record()))
+		}
+		testutil.Equals(t, 9, i, "wrong number of records")
+		testutil.Assert(t, !reader.Next(), "unexpected record")
+		testutil.Equals(t, nil, reader.Err())
+	}
+}
+
 func BenchmarkWAL_LogBatched(b *testing.B) {
 	dir, err := ioutil.TempDir("", "bench_logbatch")
 	testutil.Ok(b, err)