mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-02-02 08:31:11 -08:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Fixed race between compact (gc, populate) and head append causing unknown symbol error. (#7560)

Browse source 
* Fixed race between compact (gc, populate) and head append causing unknown symbol error.

Fixes https://github.com/prometheus/prometheus/issues/7373

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

* Addressed comments.

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>
This commit is contained in:
Bartlomiej Plotka 2020-07-14 09:36:22 +01:00 committed by GitHub
parent d77b56e88e
commit 823b218e1b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 9 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
tsdb/compact.go
View file

@ -656,7 +656,9 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
defer func() { defer func() {
var merr tsdb_errors.MultiError var merr tsdb_errors.MultiError
merr.Add(err) merr.Add(err)
merr.Add(closeAll(closers)) if cerr := closeAll(closers); cerr != nil {
merr.Add(errors.Wrap(cerr, "close"))
}
err = merr.Err() err = merr.Err()
c.metrics.populatingBlocks.Set(0) c.metrics.populatingBlocks.Set(0)
}() }()
@ -708,7 +710,6 @@ func (c *LeveledCompactor) populateBlock(blocks []BlockReader, meta *BlockMeta,
s := newCompactionSeriesSet(indexr, chunkr, tombsr, all) s := newCompactionSeriesSet(indexr, chunkr, tombsr, all)
syms := indexr.Symbols() syms := indexr.Symbols()
if i == 0 { if i == 0 {
set = s set = s
symbols = syms symbols = syms

14
tsdb/head.go
View file

@ -1104,6 +1104,7 @@ func (a *headAppender) Add(lset labels.Labels, t int64, v float64) (uint64, erro
if err != nil { if err != nil {
return 0, err return 0, err
} }
if created { if created {
a.series = append(a.series, record.RefSeries{ a.series = append(a.series, record.RefSeries{
Ref: s.ref, Ref: s.ref,
@ -1318,9 +1319,12 @@ func (h *Head) gc() {
} }
// Rebuild symbols and label value indices from what is left in the postings terms. // Rebuild symbols and label value indices from what is left in the postings terms.
// symMtx ensures that append of symbols and postings is disabled for rebuild time.
h.symMtx.Lock()
defer h.symMtx.Unlock()
symbols := make(map[string]struct{}, len(h.symbols)) symbols := make(map[string]struct{}, len(h.symbols))
values := make(map[string]stringset, len(h.values)) values := make(map[string]stringset, len(h.values))
if err := h.postings.Iter(func(t labels.Label, _ index.Postings) error { if err := h.postings.Iter(func(t labels.Label, _ index.Postings) error {
symbols[t.Name] = struct{}{} symbols[t.Name] = struct{}{}
symbols[t.Value] = struct{}{} symbols[t.Value] = struct{}{}
@ -1336,13 +1340,8 @@ func (h *Head) gc() {
// This should never happen, as the iteration function only returns nil. // This should never happen, as the iteration function only returns nil.
panic(err) panic(err)
} }
h.symMtx.Lock()
h.symbols = symbols h.symbols = symbols
h.values = values h.values = values
h.symMtx.Unlock()
} }
// Tombstones returns a new reader over the head's tombstones // Tombstones returns a new reader over the head's tombstones
@ -1692,8 +1691,6 @@ func (h *Head) getOrCreateWithID(id, hash uint64, lset labels.Labels) (*memSerie
h.metrics.seriesCreated.Inc() h.metrics.seriesCreated.Inc()
atomic.AddUint64(&h.numSeries, 1) atomic.AddUint64(&h.numSeries, 1)
h.postings.Add(id, lset)
h.symMtx.Lock() h.symMtx.Lock()
defer h.symMtx.Unlock() defer h.symMtx.Unlock()
@ -1709,6 +1706,7 @@ func (h *Head) getOrCreateWithID(id, hash uint64, lset labels.Labels) (*memSerie
h.symbols[l.Value] = struct{}{} h.symbols[l.Value] = struct{}{}
} }
h.postings.Add(id, lset)
return s, true, nil return s, true, nil
} }