Repository: prometheus/tsdb
Pull request #601 (2019-05-09 22:50:54)

Export the current segment index as a metic.
Add a metric for the current segment index that TSDB will use for the WAL. Useful for looking into WAL watcher/tailer issues.

Signed-off-by: Callum Styan <callumstyan@gmail.com>

krasi-georgiev: The current index is always the one with the highest number, why do you need a metric for that?
brian-brazil: :+1: 
krasi-georgiev: and if we decide to add this please don't forget to add a test for that metric. 
brian-brazil: We don't usually add tests for metrics.
krasi-georgiev: I prefer if we have these as we had  few broken ones here and in Prometheus with some refactoring. 
krasi-georgiev: look at the wal custom size test on how to easily trigger creation of a new wal file so you can check the metric is incremented
cstyan: @krasi-georgiev 

> The current segment is always the one with the highest number, why do you need a metric for that?

The WAL watcher code exports a metric for what segment it's currently reading from. There's a WAL watcher per remote write queue. If there's an issue with one of the remote write queue's it's useful to know whether the segment that's currently being read is the segment TSDB is currently writing to, or how far behind the queue is.

Patch:
diff --git a/Makefile.common b/Makefile.common
index 4f18ea5..c7f9ea6 100644
--- a/Makefile.common
+++ b/Makefile.common
@@ -69,7 +69,7 @@ else
 	GO_BUILD_PLATFORM ?= $(GOHOSTOS)-$(GOHOSTARCH)
 endif
 
-PROMU_VERSION ?= 0.3.0
+PROMU_VERSION ?= 0.4.0
 PROMU_URL     := https://github.com/prometheus/promu/releases/download/v$(PROMU_VERSION)/promu-$(PROMU_VERSION).$(GO_BUILD_PLATFORM).tar.gz
 
 GOLANGCI_LINT :=
diff --git a/head.go b/head.go
index 30cbe61..92619a6 100644
--- a/head.go
+++ b/head.go
@@ -421,6 +421,10 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 					if itv.Maxt < h.minValidTime {
 						continue
 					}
+					if m := h.series.getByID(s.ref); m == nil {
+						unknownRefs++
+						continue
+					}
 					allStones.addInterval(s.ref, itv)
 				}
 			}
diff --git a/index/postings.go b/index/postings.go
index cdad939..bb7b583 100644
--- a/index/postings.go
+++ b/index/postings.go
@@ -404,7 +404,6 @@ func (h *postingsHeap) Pop() interface{} {
 type mergedPostings struct {
 	h          postingsHeap
 	initilized bool
-	heaped     bool
 	cur        uint64
 	err        error
 }
@@ -434,12 +433,9 @@ func (it *mergedPostings) Next() bool {
 		return false
 	}
 
-	if !it.heaped {
-		heap.Init(&it.h)
-		it.heaped = true
-	}
 	// The user must issue an initial Next.
 	if !it.initilized {
+		heap.Init(&it.h)
 		it.cur = it.h[0].At()
 		it.initilized = true
 		return true
@@ -477,33 +473,24 @@ func (it *mergedPostings) Seek(id uint64) bool {
 			return false
 		}
 	}
-	if it.cur >= id {
-		return true
-	}
-	// Heapifying when there is lots of Seeks is inefficient,
-	// mark to be re-heapified on the Next() call.
-	it.heaped = false
-	lowest := ^uint64(0)
-	n := 0
-	for _, i := range it.h {
-		if i.Seek(id) {
-			it.h[n] = i
-			n++
-			if i.At() < lowest {
-				lowest = i.At()
+	for it.cur < id {
+		cur := it.h[0]
+		if !cur.Seek(id) {
+			heap.Pop(&it.h)
+			if cur.Err() != nil {
+				it.err = cur.Err()
+				return false
 			}
-		} else {
-			if i.Err() != nil {
-				it.err = i.Err()
+			if it.h.Len() == 0 {
 				return false
 			}
+		} else {
+			// Value of top of heap has changed, re-heapify.
+			heap.Fix(&it.h, 0)
 		}
+
+		it.cur = it.h[0].At()
 	}
-	it.h = it.h[:n]
-	if len(it.h) == 0 {
-		return false
-	}
-	it.cur = lowest
 	return true
 }
 
diff --git a/querier.go b/querier.go
index b7116c2..9d99de0 100644
--- a/querier.go
+++ b/querier.go
@@ -332,15 +332,6 @@ func PostingsForMatchers(ix IndexReader, ms ...labels.Matcher) (index.Postings,
 	it := index.Intersect(its...)
 
 	for _, n := range notIts {
-		if _, ok := n.(*index.ListPostings); !ok {
-			// Best to pre-calculate the merged lists via next rather than have a ton
-			// of seeks in Without.
-			pl, err := index.ExpandPostings(n)
-			if err != nil {
-				return nil, err
-			}
-			n = index.NewListPostings(pl)
-		}
 		it = index.Without(it, n)
 	}
 
diff --git a/wal/wal.go b/wal/wal.go
index a51d436..46504f0 100644
--- a/wal/wal.go
+++ b/wal/wal.go
@@ -171,6 +171,7 @@ type WAL struct {
 	pageCompletions prometheus.Counter
 	truncateFail    prometheus.Counter
 	truncateTotal   prometheus.Counter
+	currentSegment  prometheus.Gauge
 }
 
 // New returns a new WAL over the given directory.
@@ -218,8 +219,12 @@ func NewSize(logger log.Logger, reg prometheus.Registerer, dir string, segmentSi
 		Name: "prometheus_tsdb_wal_truncations_total",
 		Help: "Total number of WAL truncations attempted.",
 	})
+	w.currentSegment = prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "prometheus_tsdb_wal_segment_current",
+		Help: "WAL segment index that TSDB is currently writing to.",
+	})
 	if reg != nil {
-		reg.MustRegister(w.fsyncDuration, w.pageFlushes, w.pageCompletions, w.truncateFail, w.truncateTotal)
+		reg.MustRegister(w.fsyncDuration, w.pageFlushes, w.pageCompletions, w.truncateFail, w.truncateTotal, w.currentSegment)
 	}
 
 	_, j, err := w.Segments()
@@ -413,7 +418,7 @@ func (w *WAL) setSegment(segment *Segment) error {
 		return err
 	}
 	w.donePages = int(stat.Size() / pageSize)
-
+	w.currentSegment.Set(float64(segment.Index()))
 	return nil
 }
 
@@ -426,10 +431,10 @@ func (w *WAL) flushPage(clear bool) error {
 	p := w.page
 	clear = clear || p.full()
 
-	// No more data will fit into the page. Enqueue and clear it.
+	// No more data will fit into the page or an implicit clear.
+	// Enqueue and clear it.
 	if clear {
 		p.alloc = pageSize // Write till end of page.
-		w.pageCompletions.Inc()
 	}
 	n, err := w.segment.Write(p.buf[p.flushed:p.alloc])
 	if err != nil {
@@ -445,6 +450,7 @@ func (w *WAL) flushPage(clear bool) error {
 		p.alloc = 0
 		p.flushed = 0
 		w.donePages++
+		w.pageCompletions.Inc()
 	}
 	return nil
 }
@@ -495,10 +501,18 @@ func (w *WAL) Log(recs ...[]byte) error {
 	return nil
 }
 
-// log writes rec to the log and forces a flush of the current page if its
-// the final record of a batch, the record is bigger than the page size or
-// the current page is full.
+// log writes rec to the log and forces a flush of the current page if:
+// - the final record of a batch
+// - the record is bigger than the page size
+// - the current page is full.
 func (w *WAL) log(rec []byte, final bool) error {
+	// When the last page flush failed the page will remain full.
+	// When the page is full, need to flush it before trying to add more records to it.
+	if w.page.full() {
+		if err := w.flushPage(true); err != nil {
+			return err
+		}
+	}
 	// If the record is too big to fit within the active page in the current
 	// segment, terminate the active segment and advance to the next one.
 	// This ensures that records do not cross segment boundaries.


Test patch:
diff --git a/head_bench_test.go b/head_bench_test.go
index 68505ea..0b0391e 100644
--- a/head_bench_test.go
+++ b/head_bench_test.go
@@ -56,20 +56,21 @@ func BenchmarkHeadPostingForMatchers(b *testing.B) {
 		testutil.Ok(b, h.Close())
 	}()
 
-	var hash uint64
+	var ref uint64
+
+	addSeries := func(l labels.Labels) {
+		ref++
+		h.getOrCreateWithID(ref, l.Hash(), l)
+	}
+
 	for n := 0; n < 10; n++ {
 		for i := 0; i < 100000; i++ {
-			h.getOrCreate(hash, labels.FromStrings("i", strconv.Itoa(i), "n", strconv.Itoa(i), "j", "foo"))
-			hash++
+			addSeries(labels.FromStrings("i", strconv.Itoa(i), "n", strconv.Itoa(n), "j", "foo"))
 			// Have some series that won't be matched, to properly test inverted matches.
-			h.getOrCreate(hash, labels.FromStrings("i", strconv.Itoa(i), "n", strconv.Itoa(i), "j", "bar"))
-			hash++
-			h.getOrCreate(hash, labels.FromStrings("i", strconv.Itoa(i), "n", "0_"+strconv.Itoa(i), "j", "bar"))
-			hash++
-			h.getOrCreate(hash, labels.FromStrings("i", strconv.Itoa(i), "n", "1_"+strconv.Itoa(i), "j", "bar"))
-			hash++
-			h.getOrCreate(hash, labels.FromStrings("i", strconv.Itoa(i), "n", "2_"+strconv.Itoa(i), "j", "bar"))
-			hash++
+			addSeries(labels.FromStrings("i", strconv.Itoa(i), "n", strconv.Itoa(n), "j", "bar"))
+			addSeries(labels.FromStrings("i", strconv.Itoa(i), "n", "0_"+strconv.Itoa(n), "j", "bar"))
+			addSeries(labels.FromStrings("i", strconv.Itoa(i), "n", "1_"+strconv.Itoa(n), "j", "bar"))
+			addSeries(labels.FromStrings("i", strconv.Itoa(i), "n", "2_"+strconv.Itoa(n), "j", "foo"))
 		}
 	}
 
@@ -100,6 +101,7 @@ func BenchmarkHeadPostingForMatchers(b *testing.B) {
 		{`i!=""`, []labels.Matcher{iNotEmpty}},
 		{`n="1",i=~".*",j="foo"`, []labels.Matcher{n1, iStar, jFoo}},
 		{`n="1",i=~".*",i!="2",j="foo"`, []labels.Matcher{n1, iStar, iNot2, jFoo}},
+		{`n="1",i!=""`, []labels.Matcher{n1, iNotEmpty}},
 		{`n="1",i!="",j="foo"`, []labels.Matcher{n1, iNotEmpty, jFoo}},
 		{`n="1",i=~".+",j="foo"`, []labels.Matcher{n1, iPlus, jFoo}},
 		{`n="1",i=~"1.+",j="foo"`, []labels.Matcher{n1, i1Plus, jFoo}},
diff --git a/head_test.go b/head_test.go
index 9e78b6b..79092dd 100644
--- a/head_test.go
+++ b/head_test.go
@@ -114,6 +114,9 @@ func TestHead_ReadWAL(t *testing.T) {
 			{Ref: 10, T: 101, V: 5},
 			{Ref: 50, T: 101, V: 6},
 		},
+		[]Stone{
+			{ref: 0, intervals: []Interval{{Mint: 99, Maxt: 101}}},
+		},
 	}
 	dir, err := ioutil.TempDir("", "test_read_wal")
 	testutil.Ok(t, err)
diff --git a/wal/wal_test.go b/wal/wal_test.go
index 391c4b6..5363ebf 100644
--- a/wal/wal_test.go
+++ b/wal/wal_test.go
@@ -23,6 +23,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	client_testutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/tsdb/testutil"
 )
 
@@ -318,6 +319,35 @@ func TestClose(t *testing.T) {
 	testutil.NotOk(t, w.Close())
 }
 
+func TestSegmentMetric(t *testing.T) {
+	var (
+		segmentSize = pageSize
+		recordSize  = (pageSize / 2) - recordHeaderSize
+	)
+
+	dir, err := ioutil.TempDir("", "segment_metric")
+	testutil.Ok(t, err)
+	defer func() {
+		testutil.Ok(t, os.RemoveAll(dir))
+	}()
+	w, err := NewSize(nil, nil, dir, segmentSize)
+	testutil.Ok(t, err)
+
+	initialSegment := client_testutil.ToFloat64(w.currentSegment)
+
+	// Write 3 records, each of which is half the segment size, meaning we should rotate to the next segment.
+	for i := 0; i < 3; i++ {
+		buf := make([]byte, recordSize)
+		_, err := rand.Read(buf)
+		testutil.Ok(t, err)
+
+		err = w.Log(buf)
+		testutil.Ok(t, err)
+	}
+	testutil.Assert(t, client_testutil.ToFloat64(w.currentSegment) == initialSegment+1, "segment metric did not increment after segment rotation")
+	testutil.Ok(t, w.Close())
+}
+
 func BenchmarkWAL_LogBatched(b *testing.B) {
 	dir, err := ioutil.TempDir("", "bench_logbatch")
 	testutil.Ok(b, err)