Log series on rollback

Signed-off-by: Fabian Reinartz <freinartz@google.com>
This commit is contained in:
Fabian Reinartz 2018-06-28 09:04:07 -04:00
parent 16727277e4
commit ea607b9fc3
2 changed files with 37 additions and 2 deletions

View file

@ -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

View file

@ -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")}})
}