tsdb/agent: ensure that new series get written to WAL on rollback (#12592)
If a new series is introduced in a storage.Appender instance, that series should be written to the WAL once the storage.Appender is closed, even on Rollback. Previously, new series would only be written to the WAL when calling Commit. However, because the series is stored in memory regardless, subsequent calls to Commit may write samples to the WAL which reference a series ID which that was never written. Related to #11589. It's likely that this fix also resolves this issue, but we need more testing from users to see if the problem persists after this fix; there may be more cases where samples get written to the WAL in Prometheus Agent mode without the corresponding series record. Signed-off-by: Robert Fratto <robertfratto@gmail.com>
This commit is contained in:
parent
ed1b307bca
commit
886945cda7
|
@ -985,11 +985,25 @@ func (a *appender) UpdateMetadata(storage.SeriesRef, labels.Labels, metadata.Met
|
|||
|
||||
// Commit submits the collected samples and purges the batch.
|
||||
func (a *appender) Commit() error {
|
||||
if err := a.log(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
a.clearData()
|
||||
a.appenderPool.Put(a)
|
||||
return nil
|
||||
}
|
||||
|
||||
// log logs all pending data to the WAL.
|
||||
func (a *appender) log() error {
|
||||
a.mtx.RLock()
|
||||
defer a.mtx.RUnlock()
|
||||
|
||||
var encoder record.Encoder
|
||||
buf := a.bufPool.Get().([]byte)
|
||||
defer func() {
|
||||
a.bufPool.Put(buf) //nolint:staticcheck
|
||||
}()
|
||||
|
||||
if len(a.pendingSeries) > 0 {
|
||||
buf = encoder.Series(a.pendingSeries, buf)
|
||||
|
@ -1051,12 +1065,11 @@ func (a *appender) Commit() error {
|
|||
}
|
||||
}
|
||||
|
||||
//nolint:staticcheck
|
||||
a.bufPool.Put(buf)
|
||||
return a.Rollback()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *appender) Rollback() error {
|
||||
// clearData clears all pending data.
|
||||
func (a *appender) clearData() {
|
||||
a.pendingSeries = a.pendingSeries[:0]
|
||||
a.pendingSamples = a.pendingSamples[:0]
|
||||
a.pendingHistograms = a.pendingHistograms[:0]
|
||||
|
@ -1065,6 +1078,39 @@ func (a *appender) Rollback() error {
|
|||
a.sampleSeries = a.sampleSeries[:0]
|
||||
a.histogramSeries = a.histogramSeries[:0]
|
||||
a.floatHistogramSeries = a.floatHistogramSeries[:0]
|
||||
}
|
||||
|
||||
func (a *appender) Rollback() error {
|
||||
// Series are created in-memory regardless of rollback. This means we must
|
||||
// log them to the WAL, otherwise subsequent commits may reference a series
|
||||
// which was never written to the WAL.
|
||||
if err := a.logSeries(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
a.clearData()
|
||||
a.appenderPool.Put(a)
|
||||
return nil
|
||||
}
|
||||
|
||||
// logSeries logs only pending series records to the WAL.
|
||||
func (a *appender) logSeries() error {
|
||||
a.mtx.RLock()
|
||||
defer a.mtx.RUnlock()
|
||||
|
||||
if len(a.pendingSeries) > 0 {
|
||||
buf := a.bufPool.Get().([]byte)
|
||||
defer func() {
|
||||
a.bufPool.Put(buf) //nolint:staticcheck
|
||||
}()
|
||||
|
||||
var encoder record.Encoder
|
||||
buf = encoder.Series(a.pendingSeries, buf)
|
||||
if err := a.wal.Log(buf); err != nil {
|
||||
return err
|
||||
}
|
||||
buf = buf[:0]
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
@ -333,8 +333,8 @@ func TestRollback(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// Check that the rollback ensured nothing got stored.
|
||||
require.Equal(t, 0, walSeriesCount, "series should not have been written to WAL")
|
||||
// Check that only series get stored after calling Rollback.
|
||||
require.Equal(t, numSeries*3, walSeriesCount, "series should have been written to WAL")
|
||||
require.Equal(t, 0, walSamplesCount, "samples should not have been written to WAL")
|
||||
require.Equal(t, 0, walExemplarsCount, "exemplars should not have been written to WAL")
|
||||
require.Equal(t, 0, walHistogramCount, "histograms should not have been written to WAL")
|
||||
|
|
Loading…
Reference in New Issue