Stop appending duplicate exemplars

CHANGELOG.md

@@ -5,6 +5,8 @@
 - [ENHANCEMENT] integrations-next: Add `extra_labels` to add a custom set of
   labels to integration targets. (@rfratto)
 
+- [ENHANCEMENT] The agent no longer appends duplicate exemplars. (@tpaschalis)
+
 - [BUGFIX] Fixed issue where Grafana Agent may panic if there is a very large
   WAL loading while old WALs are being deleted or the `/agent/api/v1/targets`
   endpoint is called. (@tpaschalis)

pkg/metrics/wal/series.go

@@ -84,6 +84,12 @@ func (m seriesHashmap) del(hash uint64, ref uint64) {
 	}
 }
 
+type memExemplar struct {
+	ts     int64
+	value  float64
+	labels labels.Labels
+}
+
 const (
 	// defaultStripeSize is the default number of entries to allocate in the
 	// stripeSeries hash map.
@@ -97,10 +103,11 @@ const (
 //
 // This code is copied from the Prometheus TSDB.
 type stripeSeries struct {
-	size   int
-	series []map[uint64]*memSeries
-	hashes []seriesHashmap
-	locks  []stripeLock
+	size      int
+	series    []map[uint64]*memSeries
+	hashes    []seriesHashmap
+	exemplars []map[uint64]*memExemplar
+	locks     []stripeLock
 }
 
 type stripeLock struct {
@@ -112,10 +119,11 @@ type stripeLock struct {
 func newStripeSeries() *stripeSeries {
 	stripeSize := defaultStripeSize
 	s := &stripeSeries{
-		size:   stripeSize,
-		series: make([]map[uint64]*memSeries, stripeSize),
-		hashes: make([]seriesHashmap, stripeSize),
-		locks:  make([]stripeLock, stripeSize),
+		size:      stripeSize,
+		series:    make([]map[uint64]*memSeries, stripeSize),
+		hashes:    make([]seriesHashmap, stripeSize),
+		exemplars: make([]map[uint64]*memExemplar, stripeSize),
+		locks:     make([]stripeLock, stripeSize),
 	}
 
 	for i := range s.series {
@@ -124,6 +132,9 @@ func newStripeSeries() *stripeSeries {
 	for i := range s.hashes {
 		s.hashes[i] = seriesHashmap{}
 	}
+	for i := range s.exemplars {
+		s.exemplars[i] = map[uint64]*memExemplar{}
+	}
 	return s
 }
 
@@ -171,6 +182,10 @@ func (s *stripeSeries) gc(mint int64) map[uint64]struct{} {
 			delete(s.series[i], series.ref)
 			s.hashes[j].del(seriesHash, series.ref)
 
+			// Since the series is gone, we'll also delete
+			// the latest stored exemplar.
+			delete(s.exemplars[i], series.ref)
+
 			if i != j {
 				s.locks[j].Unlock()
 			}
@@ -216,6 +231,24 @@ func (s *stripeSeries) set(hash uint64, series *memSeries) {
 	s.locks[i].Unlock()
 }
 
+func (s *stripeSeries) getLatestExemplar(id uint64) *memExemplar {
+	i := id & uint64(s.size-1)
+
+	s.locks[i].RLock()
+	exemplar := s.exemplars[i][id]
+	s.locks[i].RUnlock()
+
+	return exemplar
+}
+
+func (s *stripeSeries) setLatestExemplar(id uint64, exemplar *memExemplar) {
+	i := id & uint64(s.size-1)
+
+	s.locks[i].Lock()
+	s.exemplars[i][id] = exemplar
+	s.locks[i].Unlock()
+}
+
 func (s *stripeSeries) iterator() *stripeSeriesIterator {
 	return &stripeSeriesIterator{s}
 }

pkg/metrics/wal/wal.go

@@ -646,6 +646,14 @@ func (a *appender) AppendExemplar(ref uint64, _ labels.Labels, e exemplar.Exempl
 		}
 	}
 
+	// Check for duplicate vs last stored exemplar for this series, and discard those.
+	// Otherwise, record the current exemplar as the latest
+	prevExemplar := a.w.series.getLatestExemplar(ref)
+	if prevExemplar != nil && exemplarsEqual(*prevExemplar, e) {
+		return 0, nil
+	}
+	a.w.series.setLatestExemplar(ref, &memExemplar{ts: e.Ts, value: e.Value, labels: e.Labels})
+
 	a.exemplars = append(a.exemplars, record.RefExemplar{
 		Ref:    ref,
 		T:      e.Ts,
@@ -714,3 +722,15 @@ func (a *appender) Rollback() error {
 	a.w.appenderPool.Put(a)
 	return nil
 }
+
+func exemplarsEqual(me memExemplar, pe exemplar.Exemplar) bool {
+	if !labels.Equal(me.labels, pe.Labels) {
+		return false
+	}
+
+	if me.ts != pe.Ts {
+		return false
+	}
+
+	return me.value == pe.Value
+}

pkg/metrics/wal/wal_test.go

@@ -104,6 +104,47 @@ func TestStorage(t *testing.T) {
 	require.Equal(t, expectedExemplars, actualExemplars)
 }
 
+func TestStorage_DuplicateExemplarsIgnored(t *testing.T) {
+	walDir, err := ioutil.TempDir(os.TempDir(), "wal")
+	require.NoError(t, err)
+	defer os.RemoveAll(walDir)
+
+	s, err := NewStorage(log.NewNopLogger(), nil, walDir)
+	require.NoError(t, err)
+
+	app := s.Appender(context.Background())
+
+	sRef, err := app.Append(0, labels.Labels{{Name: "a", Value: "1"}}, 0, 0)
+	require.NoError(t, err, "should not reject valid series")
+
+	// If the Labels, Value or Timestamp are different than the last exemplar,
+	// then a new one should be appended; Otherwise, it should be skipped.
+	e := exemplar.Exemplar{Labels: labels.Labels{{Name: "a", Value: "1"}}, Value: 20, Ts: 10, HasTs: true}
+	_, _ = app.AppendExemplar(sRef, nil, e)
+	_, _ = app.AppendExemplar(sRef, nil, e)
+
+	e.Labels = labels.Labels{{Name: "b", Value: "2"}}
+	_, _ = app.AppendExemplar(sRef, nil, e)
+	_, _ = app.AppendExemplar(sRef, nil, e)
+	_, _ = app.AppendExemplar(sRef, nil, e)
+
+	e.Value = 42
+	_, _ = app.AppendExemplar(sRef, nil, e)
+	_, _ = app.AppendExemplar(sRef, nil, e)
+
+	e.Ts = 25
+	_, _ = app.AppendExemplar(sRef, nil, e)
+	_, _ = app.AppendExemplar(sRef, nil, e)
+
+	require.NoError(t, app.Commit())
+	collector := walDataCollector{}
+	replayer := walReplayer{w: &collector}
+	require.NoError(t, replayer.Replay(s.wal.Dir()))
+
+	// We had 9 calls to AppendExemplar but only 4 of those should have gotten through
+	require.Equal(t, 4, len(collector.exemplars))
+}
+
 func TestStorage_ExistingWAL(t *testing.T) {
 	walDir, err := ioutil.TempDir(os.TempDir(), "wal")
 	require.NoError(t, err)