Repository: prometheus/tsdb
Pull request #270 (2018-01-28 16:29:16)

Head: don't create stones, delete samples directly
This should close [#266](https://github.com/prometheus/tsdb/issues/266) and [https://github.com/prometheus/prometheus/issues/3728](https://github.com/prometheus/prometheus/issues/3728)

/cc @Gouthamve 

Gouthamve: Can you add a test for this?
codesome: Sorry, forgot to run the test before commit. Anyway, all tests assume that samples are not actually deleted after calling Delete(). I guess those tests need to be updated?
codesome: @Gouthamve @fabxc 
So this is what I have done to take care of `sampleBuf` while clearing tombstones. (failed test for commit 61cc663 was because of not handling this)

As far as I understand `sampleBuf` holds the last 4 samples in that series. So after clearing tombstones I just update the `sampleBuf` using the last 4 samples in the series.

Example for visualisation:
(the numbers are only t from sample and not showing v)

Current samples: `0 1 2 3 4 5 6 7 8 9`
sampleBuf: `6 7 8 9`
representing as => mem: `0 1 2 3 4 5 6 7 8 9 | 6 7 8 9`

Tombstones: `{{2,3}, {4,7}, {0,1}}`

Deleting:

`{2,3}`
mem: `0 1 4 5 6 7 8 9 | 6 7 8 9`

`{4,7}`
mem: `0 1 8 9 | 0 1 8 9`

`{0,1}`
mem: `8 9 | 0 0 8 9` (first 2 samples in `sampleBuf` are empty)

Please let me know if what I have done is correct.

PS: Test for `head.CleanTombstones()` will be added soon.
codesome: @Gouthamve, need a review on this. The last test on 092c35a looks falky, need to restart the test again.
codesome: @Gouthamve need a review. Test on 78b8cc8 looks flaky.
codesome: @Gouthamve need a review. Test on 09ad2ed looks flaky.
codesome: @Gouthamve as discussed offline, I have removed the 'tombstones' field from Head.
Gouthamve: LGTM 👍 !

@fabxc Essentially this deletes data from memory instantly and removes any series with no chunks. This is required to tackle https://github.com/prometheus/prometheus/issues/3728.

I like removing the data from memory instantly rather than calling a seperate `CleanTombstones` to do it. I'll test this under load later tonight but this is good for a review now.
fabxc: Mh, is this fixing an immediate bug? Otherwise I'd like to put it on hold until prometheus/tsdb is stable again.

Is prometheus/prometheus#3728 really an issue? On compaction the stale job labels should dropped eventually no?
codesome: This can be an enhancement, rather than just fixing the issue of labels. As its cheaper to delete samples when in memory, we will avoid some disk writes while cleaning tombstones from persistent blocks (in some cases).

Anyway, better to hold this till 2.2.0 is released.
krasi-georgiev: @fabxc do you predict  any problems with this change or should we move on with the reviews?
krasi-georgiev: @codesome would you mind resolving the conflicts and I will review it.

bare in mind the changes in https://github.com/prometheus/tsdb/pull/378

codesome: This happens to be my first PR to tsdb (also when I was just getting introduced to Prometheus), and I already found a small bug while loading WAL with this PR.

Resolving conflicts took some time, many things had changed. I will push once I clean up the code and verify once more.
krasi-georgiev: ok ping me when ready for a review.
codesome: ping @krasi-georgiev 
krasi-georgiev: thanks , I will review once we clear the other few pending PRs that were before this one.
codesome: @krasi-georgiev I will address the reviews in couple of days, caught up with university work.
krasi-georgiev: sure , ping me when ready for another review.

codesome: @krasi-georgiev ping
codesome: @krasi-georgiev as discussed over chat, we want to test the behaviour of the exposed APIs and not all unexposed API. 

So basically we want to remove `TestMemSeries_chunkMetas` and `TestMemSeriesReset`, and test the Querier and other exposed API to see if these don't cause any side affect.

Please correct me if I have missed something.
krasi-georgiev: correct, the last behaviour we want to test is that when we delete samples from the head something that uses the buffer should behave as expected and in this case following the stack up it is the (db *DB) Querier.
krasi-georgiev:  @codesome lets do one final push to get this merged :smiley_cat: 
codesome: Was busy with some other PRs. Will get on this soon.
codesome: @krasi-georgiev I would like to keep `TestMemSeriesReset`. Reason: When I merged this PR with master to remove conflicts, almost all delete tests failed (including `TestMemSeriesReset`), and apparently it was a bug in `reset()` itself. `TestMemSeriesReset` helped me debug it (just had to add `s.headChunk = nil`). 

So for the easy of debugging, I will keep that test and add more tests as discussed above.
krasi-georgiev: ok
codesome: ping @krasi-georgiev 
codesome: @krasi-georgiev do you think we should run benchmark on this in prometheus once to test some deletes manually on real setting too?
krasi-georgiev: time to squash for the DCO.

> @krasi-georgiev do you think we should run benchmark on this in prometheus once to test some deletes manually on real setting too?

I would say do all the manual tests you can think of locally and that should be good enough.

Patch:
diff --git a/head.go b/head.go
index cbc8661..637e847 100644
--- a/head.go
+++ b/head.go
@@ -48,6 +48,10 @@ var (
 	// ErrOutOfBounds is returned if an appended sample is out of the
 	// writable time range.
 	ErrOutOfBounds = errors.New("out of bounds")
+
+	// emptyTombstoneReader is a no-op Tombstone Reader.
+	// This is used by head to satisfy the Tombstones() function call.
+	emptyTombstoneReader = newMemTombstones()
 )
 
 // Head handles reads and writes of time series data within a time window.
@@ -71,8 +75,6 @@ type Head struct {
 	values  map[string]stringset // label names to possible values
 
 	postings *index.MemPostings // postings lists for terms
-
-	tombstones *memTombstones
 }
 
 type headMetrics struct {
@@ -231,7 +233,6 @@ func NewHead(r prometheus.Registerer, l log.Logger, wal *wal.WAL, chunkRange int
 		values:     map[string]stringset{},
 		symbols:    map[string]struct{}{},
 		postings:   index.NewUnorderedMemPostings(),
-		tombstones: newMemTombstones(),
 	}
 	h.metrics = newHeadMetrics(h, r)
 
@@ -334,12 +335,14 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 	}
 
 	var (
-		dec     RecordDecoder
-		series  []RefSeries
-		samples []RefSample
-		tstones []Stone
-		err     error
+		dec       RecordDecoder
+		series    []RefSeries
+		samples   []RefSample
+		tstones   []Stone
+		allStones = newMemTombstones()
+		err       error
 	)
+	defer allStones.Close()
 	for r.Next() {
 		series, samples, tstones = series[:0], samples[:0], tstones[:0]
 		rec := r.Record()
@@ -413,7 +416,7 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 					if itv.Maxt < h.minValidTime {
 						continue
 					}
-					h.tombstones.addInterval(s.ref, itv)
+					allStones.addInterval(s.ref, itv)
 				}
 			}
 		default:
@@ -436,6 +439,12 @@ func (h *Head) loadWAL(r *wal.Reader) error {
 	}
 	wg.Wait()
 
+	if err := allStones.Iter(func(ref uint64, dranges Intervals) error {
+		return h.chunkRewrite(ref, dranges)
+	}); err != nil {
+		return errors.Wrap(r.Err(), "deleting samples from tombstones")
+	}
+
 	if unknownRefs > 0 {
 		level.Warn(h.logger).Log("msg", "unknown series references", "count", unknownRefs)
 	}
@@ -604,7 +613,7 @@ func (h *rangeHead) Chunks() (ChunkReader, error) {
 }
 
 func (h *rangeHead) Tombstones() (TombstoneReader, error) {
-	return h.head.tombstones, nil
+	return emptyTombstoneReader, nil
 }
 
 // initAppender is a helper to initialize the time bounds of the head
@@ -849,7 +858,7 @@ func (h *Head) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 	}
 
 	var stones []Stone
-
+	dirty := false
 	for p.Next() {
 		series := h.series.getByID(p.At())
 
@@ -859,22 +868,61 @@ func (h *Head) Delete(mint, maxt int64, ms ...labels.Matcher) error {
 		}
 		// Delete only until the current values and not beyond.
 		t0, t1 = clampInterval(mint, maxt, t0, t1)
-		stones = append(stones, Stone{p.At(), Intervals{{t0, t1}}})
+		if h.wal != nil {
+			stones = append(stones, Stone{p.At(), Intervals{{t0, t1}}})
+		}
+		if err := h.chunkRewrite(p.At(), Intervals{{t0, t1}}); err != nil {
+			return errors.Wrap(err, "delete samples")
+		}
+		dirty = true
 	}
-
 	if p.Err() != nil {
 		return p.Err()
 	}
 	var enc RecordEncoder
-
 	if h.wal != nil {
+		// Although we don't store the stones in the head
+		// we need to write them  to the WAL to mark these as deleted
+		// after a restart while loeading the WAL.
 		if err := h.wal.Log(enc.Tombstones(stones, nil)); err != nil {
 			return err
 		}
 	}
-	for _, s := range stones {
-		h.tombstones.addInterval(s.ref, s.intervals[0])
+	if dirty {
+		h.gc()
+	}
+
+	return nil
+}
+
+// chunkRewrite re-writes the chunks which overlaps with deleted ranges
+// and removes the samples in the deleted ranges.
+// Chunks is deleted if no samples are left at the end.
+func (h *Head) chunkRewrite(ref uint64, dranges Intervals) (err error) {
+	if len(dranges) == 0 {
+		return nil
+	}
+
+	ms := h.series.getByID(ref)
+	ms.Lock()
+	defer ms.Unlock()
+	if len(ms.chunks) == 0 {
+		return nil
+	}
+
+	metas := ms.chunksMetas()
+	mint, maxt := metas[0].MinTime, metas[len(metas)-1].MaxTime
+	it := newChunkSeriesIterator(metas, dranges, mint, maxt)
+
+	ms.reset()
+	for it.Next() {
+		t, v := it.At()
+		ok, _ := ms.append(t, v)
+		if !ok {
+			level.Warn(h.logger).Log("msg", "failed to add sample during delete")
+		}
 	}
+
 	return nil
 }
 
@@ -926,7 +974,7 @@ func (h *Head) gc() {
 
 // Tombstones returns a new reader over the head's tombstones
 func (h *Head) Tombstones() (TombstoneReader, error) {
-	return h.tombstones, nil
+	return emptyTombstoneReader, nil
 }
 
 // Index returns an IndexReader against the block.
@@ -1406,6 +1454,16 @@ type memSeries struct {
 	app chunkenc.Appender // Current appender for the chunk.
 }
 
+func newMemSeries(lset labels.Labels, id uint64, chunkRange int64) *memSeries {
+	s := &memSeries{
+		lset:       lset,
+		ref:        id,
+		chunkRange: chunkRange,
+		nextAt:     math.MinInt64,
+	}
+	return s
+}
+
 func (s *memSeries) minTime() int64 {
 	if len(s.chunks) == 0 {
 		return math.MinInt64
@@ -1442,14 +1500,24 @@ func (s *memSeries) cut(mint int64) *memChunk {
 	return c
 }
 
-func newMemSeries(lset labels.Labels, id uint64, chunkRange int64) *memSeries {
-	s := &memSeries{
-		lset:       lset,
-		ref:        id,
-		chunkRange: chunkRange,
-		nextAt:     math.MinInt64,
+func (s *memSeries) chunksMetas() []chunks.Meta {
+	metas := make([]chunks.Meta, 0, len(s.chunks))
+	for _, chk := range s.chunks {
+		metas = append(metas, chunks.Meta{Chunk: chk.chunk, MinTime: chk.minTime, MaxTime: chk.maxTime})
 	}
-	return s
+	return metas
+}
+
+// reset re-initialises all the variable in the memSeries except 'lset', 'ref',
+// and 'chunkRange', like how it would appear after 'newMemSeries(...)'.
+func (s *memSeries) reset() {
+	s.chunks = nil
+	s.headChunk = nil
+	s.firstChunkID = 0
+	s.nextAt = math.MinInt64
+	s.sampleBuf = [4]sample{}
+	s.pendingCommit = false
+	s.app = nil
 }
 
 // appendable checks whether the given sample is valid for appending to the series.


Test patch:
diff --git a/db_test.go b/db_test.go
index f727a79..06f0141 100644
--- a/db_test.go
+++ b/db_test.go
@@ -213,31 +213,48 @@ func TestDBAppenderAddRef(t *testing.T) {
 func TestDeleteSimple(t *testing.T) {
 	numSamples := int64(10)
 
-	db, close := openTestDB(t, nil)
-	defer close()
-	defer db.Close()
-
-	app := db.Appender()
-
-	smpls := make([]float64, numSamples)
-	for i := int64(0); i < numSamples; i++ {
-		smpls[i] = rand.Float64()
-		app.Add(labels.Labels{{"a", "b"}}, i, smpls[i])
-	}
-
-	testutil.Ok(t, app.Commit())
 	cases := []struct {
 		intervals Intervals
 		remaint   []int64
 	}{
+		{
+			intervals: Intervals{{0, 3}},
+			remaint:   []int64{4, 5, 6, 7, 8, 9},
+		},
+		{
+			intervals: Intervals{{1, 3}},
+			remaint:   []int64{0, 4, 5, 6, 7, 8, 9},
+		},
 		{
 			intervals: Intervals{{1, 3}, {4, 7}},
 			remaint:   []int64{0, 8, 9},
 		},
+		{
+			intervals: Intervals{{1, 3}, {4, 700}},
+			remaint:   []int64{0},
+		},
+		{ // This case is to ensure that labels and symbols are deleted.
+			intervals: Intervals{{0, 9}},
+			remaint:   []int64{},
+		},
 	}
 
 Outer:
 	for _, c := range cases {
+		db, close := openTestDB(t, nil)
+		defer close()
+		defer db.Close()
+
+		app := db.Appender()
+
+		smpls := make([]float64, numSamples)
+		for i := int64(0); i < numSamples; i++ {
+			smpls[i] = rand.Float64()
+			app.Add(labels.Labels{{"a", "b"}}, i, smpls[i])
+		}
+
+		testutil.Ok(t, app.Commit())
+
 		// TODO(gouthamve): Reset the tombstones somehow.
 		// Delete the ranges.
 		for _, r := range c.intervals {
@@ -260,9 +277,20 @@ Outer:
 			newSeries(map[string]string{"a": "b"}, expSamples),
 		})
 
+		lns, err := q.LabelNames()
+		testutil.Ok(t, err)
+		lvs, err := q.LabelValues("a")
+		testutil.Ok(t, err)
 		if len(expSamples) == 0 {
+			testutil.Equals(t, 0, len(lns))
+			testutil.Equals(t, 0, len(lvs))
 			testutil.Assert(t, res.Next() == false, "")
 			continue
+		} else {
+			testutil.Equals(t, 1, len(lns))
+			testutil.Equals(t, 1, len(lvs))
+			testutil.Equals(t, "a", lns[0])
+			testutil.Equals(t, "b", lvs[0])
 		}
 
 		for {
diff --git a/head_test.go b/head_test.go
index 1378183..106c4a4 100644
--- a/head_test.go
+++ b/head_test.go
@@ -18,6 +18,7 @@ import (
 	"math"
 	"math/rand"
 	"os"
+	"path"
 	"path/filepath"
 	"sort"
 	"testing"
@@ -296,94 +297,158 @@ func TestHeadDeleteSeriesWithoutSamples(t *testing.T) {
 }
 
 func TestHeadDeleteSimple(t *testing.T) {
-	numSamples := int64(10)
-
-	head, err := NewHead(nil, nil, nil, 1000)
-	testutil.Ok(t, err)
-	defer head.Close()
-
-	app := head.Appender()
-
-	smpls := make([]float64, numSamples)
-	for i := int64(0); i < numSamples; i++ {
-		smpls[i] = rand.Float64()
-		app.Add(labels.Labels{{"a", "b"}}, i, smpls[i])
+	buildSmpls := func(s []int64) []sample {
+		ss := make([]sample, 0, len(s))
+		for _, t := range s {
+			ss = append(ss, sample{t: t, v: float64(t)})
+		}
+		return ss
 	}
+	smplsAll := buildSmpls([]int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9})
+	lblDefault := labels.Label{"a", "b"}
 
-	testutil.Ok(t, app.Commit())
 	cases := []struct {
-		intervals Intervals
-		remaint   []int64
+		dranges  Intervals
+		smplsExp []sample
 	}{
 		{
-			intervals: Intervals{{0, 3}},
-			remaint:   []int64{4, 5, 6, 7, 8, 9},
+			dranges:  Intervals{{0, 3}},
+			smplsExp: buildSmpls([]int64{4, 5, 6, 7, 8, 9}),
 		},
 		{
-			intervals: Intervals{{1, 3}},
-			remaint:   []int64{0, 4, 5, 6, 7, 8, 9},
+			dranges:  Intervals{{1, 3}},
+			smplsExp: buildSmpls([]int64{0, 4, 5, 6, 7, 8, 9}),
 		},
 		{
-			intervals: Intervals{{1, 3}, {4, 7}},
-			remaint:   []int64{0, 8, 9},
+			dranges:  Intervals{{1, 3}, {4, 7}},
+			smplsExp: buildSmpls([]int64{0, 8, 9}),
 		},
 		{
-			intervals: Intervals{{1, 3}, {4, 700}},
-			remaint:   []int64{0},
+			dranges:  Intervals{{1, 3}, {4, 700}},
+			smplsExp: buildSmpls([]int64{0}),
 		},
-		{
-			intervals: Intervals{{0, 9}},
-			remaint:   []int64{},
+		{ // This case is to ensure that labels and symbols are deleted.
+			dranges:  Intervals{{0, 9}},
+			smplsExp: buildSmpls([]int64{}),
 		},
 	}
 
 Outer:
 	for _, c := range cases {
-		// Reset the tombstones.
-		head.tombstones = newMemTombstones()
+		dir, err := ioutil.TempDir("", "test_wal_reload")
+		testutil.Ok(t, err)
+		defer os.RemoveAll(dir)
+
+		w, err := wal.New(nil, nil, path.Join(dir, "wal"))
+		testutil.Ok(t, err)
+
+		head, err := NewHead(nil, nil, w, 1000)
+		testutil.Ok(t, err)
+
+		app := head.Appender()
+		for _, smpl := range smplsAll {
+			_, err = app.Add(labels.Labels{lblDefault}, smpl.t, smpl.v)
+			testutil.Ok(t, err)
+
+		}
+		testutil.Ok(t, app.Commit())
 
 		// Delete the ranges.
-		for _, r := range c.intervals {
-			testutil.Ok(t, head.Delete(r.Mint, r.Maxt, labels.NewEqualMatcher("a", "b")))
+		for _, r := range c.dranges {
+			testutil.Ok(t, head.Delete(r.Mint, r.Maxt, labels.NewEqualMatcher(lblDefault.Name, lblDefault.Value)))
 		}
 
-		// Compare the result.
-		q, err := NewBlockQuerier(head, head.MinTime(), head.MaxTime())
+		// Compare the samples for both heads - before and after the reload.
+		reloadedW, err := wal.New(nil, nil, w.Dir()) // Use a new wal to ensure deleted samples are gone even after a reload.
 		testutil.Ok(t, err)
-		res, err := q.Select(labels.NewEqualMatcher("a", "b"))
+		reloadedHead, err := NewHead(nil, nil, reloadedW, 1000)
 		testutil.Ok(t, err)
+		testutil.Ok(t, reloadedHead.Init(0))
+		for _, h := range []*Head{head, reloadedHead} {
+			indexr, err := h.Index()
+			testutil.Ok(t, err)
+			// Use an emptyTombstoneReader explicitly to get all the samples.
+			css, err := LookupChunkSeries(indexr, emptyTombstoneReader, labels.NewEqualMatcher(lblDefault.Name, lblDefault.Value))
+			testutil.Ok(t, err)
 
-		expSamples := make([]tsdbutil.Sample, 0, len(c.remaint))
-		for _, ts := range c.remaint {
-			expSamples = append(expSamples, sample{ts, smpls[ts]})
+			// Getting the actual samples.
+			actSamples := make([]sample, 0)
+			for css.Next() {
+				lblsAct, chkMetas, intv := css.At()
+				testutil.Equals(t, labels.Labels{lblDefault}, lblsAct)
+				testutil.Equals(t, 0, len(intv))
+
+				chunkr, err := h.Chunks()
+				testutil.Ok(t, err)
+				for _, meta := range chkMetas {
+					chk, err := chunkr.Chunk(meta.Ref)
+					testutil.Ok(t, err)
+					ii := chk.Iterator()
+					for ii.Next() {
+						t, v := ii.At()
+						actSamples = append(actSamples, sample{t: t, v: v})
+					}
+				}
+			}
+
+			testutil.Ok(t, css.Err())
+			testutil.Equals(t, c.smplsExp, actSamples)
 		}
 
-		expss := newMockSeriesSet([]Series{
-			newSeries(map[string]string{"a": "b"}, expSamples),
+		// Compare the query results for both heads - before and after the reload.
+		expSeriesSet := newMockSeriesSet([]Series{
+			newSeries(map[string]string{lblDefault.Name: lblDefault.Value}, func() []tsdbutil.Sample {
+				ss := make([]tsdbutil.Sample, 0, len(c.smplsExp))
+				for _, s := range c.smplsExp {
+					ss = append(ss, s)
+				}
+				return ss
+			}(),
+			),
 		})
+		for _, h := range []*Head{head, reloadedHead} {
+			q, err := NewBlockQuerier(h, h.MinTime(), h.MaxTime())
+			testutil.Ok(t, err)
+			actSeriesSet, err := q.Select(labels.NewEqualMatcher(lblDefault.Name, lblDefault.Value))
+			testutil.Ok(t, err)
 
-		if len(expSamples) == 0 {
-			testutil.Assert(t, res.Next() == false, "")
-			continue
-		}
+			lns, err := q.LabelNames()
+			testutil.Ok(t, err)
+			lvs, err := q.LabelValues(lblDefault.Name)
+			testutil.Ok(t, err)
+			// When all samples are deleted we expect that no labels should exist either.
+			if len(c.smplsExp) == 0 {
+				testutil.Equals(t, 0, len(lns))
+				testutil.Equals(t, 0, len(lvs))
+				testutil.Assert(t, actSeriesSet.Next() == false, "")
+				testutil.Ok(t, h.Close())
+				continue
+			} else {
+				testutil.Equals(t, 1, len(lns))
+				testutil.Equals(t, 1, len(lvs))
+				testutil.Equals(t, lblDefault.Name, lns[0])
+				testutil.Equals(t, lblDefault.Value, lvs[0])
+			}
 
-		for {
-			eok, rok := expss.Next(), res.Next()
-			testutil.Equals(t, eok, rok)
+			for {
+				eok, rok := expSeriesSet.Next(), actSeriesSet.Next()
+				testutil.Equals(t, eok, rok)
 
-			if !eok {
-				continue Outer
-			}
-			sexp := expss.At()
-			sres := res.At()
+				if !eok {
+					testutil.Ok(t, h.Close())
+					continue Outer
+				}
+				expSeries := expSeriesSet.At()
+				actSeries := actSeriesSet.At()
 
-			testutil.Equals(t, sexp.Labels(), sres.Labels())
+				testutil.Equals(t, expSeries.Labels(), actSeries.Labels())
 
-			smplExp, errExp := expandSeriesIterator(sexp.Iterator())
-			smplRes, errRes := expandSeriesIterator(sres.Iterator())
+				smplExp, errExp := expandSeriesIterator(expSeries.Iterator())
+				smplRes, errRes := expandSeriesIterator(actSeries.Iterator())
 
-			testutil.Equals(t, errExp, errRes)
-			testutil.Equals(t, smplExp, smplRes)
+				testutil.Equals(t, errExp, errRes)
+				testutil.Equals(t, smplExp, smplRes)
+			}
 		}
 	}
 }
@@ -524,8 +589,6 @@ func TestDelete_e2e(t *testing.T) {
 		// TODO: Add Regexp Matchers.
 	}
 	for _, del := range dels {
-		// Reset the deletes everytime.
-		hb.tombstones = newMemTombstones()
 		for _, r := range del.drange {
 			testutil.Ok(t, hb.Delete(r.Mint, r.Maxt, del.ms...))
 		}
@@ -946,4 +1009,5 @@ func TestWalRepair(t *testing.T) {
 			testutil.Equals(t, test.expRecs, actRec, "Wrong number of intact records")
 		})
 	}
+
 }
diff --git a/testutil/testutil.go b/testutil/testutil.go
index 06b9747..03784e7 100644
--- a/testutil/testutil.go
+++ b/testutil/testutil.go
@@ -66,6 +66,15 @@ func Equals(tb testing.TB, exp, act interface{}, msgAndArgs ...interface{}) {
 	}
 }
 
+// NotEquals fails the test if exp is equal to act.
+func NotEquals(tb testing.TB, exp, act interface{}) {
+	if reflect.DeepEqual(exp, act) {
+		_, file, line, _ := runtime.Caller(1)
+		fmt.Printf("\033[31m%s:%d: Expected different exp and got\n\n\texp: %#v\n\n\tgot: %#v\033[39m\n\n", filepath.Base(file), line, exp, act)
+		tb.FailNow()
+	}
+}
+
 func formatMessage(msgAndArgs []interface{}) string {
 	if len(msgAndArgs) == 0 {
 		return ""