Initialize biggest ref to existing ref when reading a segment (#676)

* initialize biggest ref to existing ref when reading a segment

* changelog

* allow new test to be less hard-coded

* explain error if test fails

* clarify changelog issue

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 - [FEATURE] Add TLS config options for tempo `remote_write`s. (@mapno)
 
+- [BUGFIX] Fix issue where replaying a WAL caused incorrect metrics to be sent
+  over remote write. (@rfratto)
+
 # v0.16.0 (2021-06-17)
 
 - [FEATURE] (beta) A Grafana Agent Operator is now available. (@rfratto)

pkg/prom/wal/wal.go

@@ -294,7 +294,7 @@ func (w *Storage) loadWAL(r *wal.Reader) (err error) {
 		}
 	}()
 
-	var biggestRef uint64 = 0
+	var biggestRef uint64 = w.ref.Load()
 
 	for d := range decoded {
 		switch v := d.(type) {

pkg/prom/wal/wal_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log"
+	"github.com/grafana/agent/pkg/util"
 	"github.com/prometheus/prometheus/pkg/exemplar"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/pkg/value"
@@ -168,6 +169,37 @@ func TestStorage_ExistingWAL(t *testing.T) {
 	require.Equal(t, expectedExemplars, actualExemplars)
 }
 
+func TestStorage_ExistingWAL_RefID(t *testing.T) {
+	l := util.TestLogger(t)
+
+	walDir, err := ioutil.TempDir(os.TempDir(), "wal")
+	require.NoError(t, err)
+	defer os.RemoveAll(walDir)
+
+	s, err := NewStorage(l, nil, walDir)
+	require.NoError(t, err)
+
+	app := s.Appender(context.Background())
+	payload := buildSeries([]string{"foo", "bar", "baz", "blerg"})
+
+	// Write all the samples
+	for _, metric := range payload {
+		metric.Write(t, app)
+	}
+	require.NoError(t, app.Commit())
+
+	// Truncate the WAL to force creation of a new segment.
+	require.NoError(t, s.Truncate(0))
+	require.NoError(t, s.Close())
+
+	// Create a new storage and see what the ref ID is initialized to.
+	s, err = NewStorage(l, nil, walDir)
+	require.NoError(t, err)
+	defer require.NoError(t, s.Close())
+
+	require.Equal(t, uint64(len(payload)), s.ref.Load(), "cached ref ID should be equal to the number of series written")
+}
+
 func TestStorage_Truncate(t *testing.T) {
 	// Same as before but now do the following:
 	// after writing all the data, forcefully create 4 more segments,