Merge pull request #356 from prometheus/logrollback

Log series on rollback

head.go

@@ -521,7 +521,8 @@ func (a *headAppender) AddFast(ref uint64, t int64, v float64) error {
 }
 
 func (a *headAppender) Commit() error {
-	defer a.Rollback()
+	defer a.head.metrics.activeAppenders.Dec()
+	defer a.head.putAppendBuffer(a.samples)
 
 	if err := a.head.wal.LogSeries(a.series); err != nil {
 		return err
@@ -565,7 +566,9 @@ func (a *headAppender) Rollback() error {
 	a.head.metrics.activeAppenders.Dec()
 	a.head.putAppendBuffer(a.samples)
 
-	return nil
+	// Series are created in the head memory regardless of rollback. Thus we have
+	// to log them to the WAL in any case.
+	return a.head.wal.LogSeries(a.series)
 }
 
 // Delete all samples in the range of [mint, maxt] for series that satisfy the given

head_test.go

@@ -47,6 +47,21 @@ type memoryWAL struct {
 	entries []interface{}
 }
 
+func (w *memoryWAL) LogSeries(s []RefSeries) error {
+	w.entries = append(w.entries, s)
+	return nil
+}
+
+func (w *memoryWAL) LogSamples(s []RefSample) error {
+	w.entries = append(w.entries, s)
+	return nil
+}
+
+func (w *memoryWAL) LogDeletes(s []Stone) error {
+	w.entries = append(w.entries, s)
+	return nil
+}
+
 func (w *memoryWAL) Reader() WALReader {
 	return w
 }
@@ -769,3 +784,20 @@ func TestGCSeriesAccess(t *testing.T) {
 	_, err = cr.Chunk(chunks[1].Ref)
 	testutil.Equals(t, ErrNotFound, err)
 }
+
+func TestHead_LogRollback(t *testing.T) {
+	w := &memoryWAL{}
+	h, err := NewHead(nil, nil, w, 1000)
+	testutil.Ok(t, err)
+
+	app := h.Appender()
+	_, err = app.Add(labels.FromStrings("a", "b"), 1, 2)
+	testutil.Ok(t, err)
+
+	testutil.Ok(t, app.Rollback())
+	testutil.Equals(t, 1, len(w.entries))
+
+	series, ok := w.entries[0].([]RefSeries)
+	testutil.Assert(t, ok, "expected series record but got %+v", w.entries[0])
+	testutil.Equals(t, series, []RefSeries{{Ref: 1, Labels: labels.FromStrings("a", "b")}})
+}