github
/
prometheus
mirror of https://github.com/prometheus/prometheus
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #15380 from bboreham/improve-loadwbl 

[BUGFIX] TSDB: Apply fixes from loadWAL to loadWBL
pull/12587/merge
Bryan Boreham 2024-11-25 17:31:49 +00:00 committed by GitHub
parent 7996a13fdd 5450e6d368
commit dd0252a774
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 15 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
tsdb/head_wal.go
View File

@ -656,32 +656,15 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
concurrency = h.opts.WALReplayConcurrency concurrency = h.opts.WALReplayConcurrency
processors = make([]wblSubsetProcessor, concurrency) processors = make([]wblSubsetProcessor, concurrency)
dec record.Decoder
shards = make([][]record.RefSample, concurrency) shards = make([][]record.RefSample, concurrency)
histogramShards = make([][]histogramRecord, concurrency) histogramShards = make([][]histogramRecord, concurrency)
decodedCh = make(chan interface{}, 10) decodedCh = make(chan interface{}, 10)
decodeErr error decodeErr error
samplesPool = sync.Pool{ samplesPool zeropool.Pool[[]record.RefSample]
New: func() interface{} { markersPool zeropool.Pool[[]record.RefMmapMarker]
return []record.RefSample{} histogramSamplesPool zeropool.Pool[[]record.RefHistogramSample]
}, floatHistogramSamplesPool zeropool.Pool[[]record.RefFloatHistogramSample]
}
markersPool = sync.Pool{
New: func() interface{} {
return []record.RefMmapMarker{}
},
}
histogramSamplesPool = sync.Pool{
New: func() interface{} {
return []record.RefHistogramSample{}
},
}
floatHistogramSamplesPool = sync.Pool{
New: func() interface{} {
return []record.RefFloatHistogramSample{}
},
}
) )
defer func() { defer func() {
@ -711,11 +694,13 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
go func() { go func() {
defer close(decodedCh) defer close(decodedCh)
var err error
dec := record.NewDecoder(syms)
for r.Next() { for r.Next() {
rec := r.Record() rec := r.Record()
switch dec.Type(rec) { switch dec.Type(rec) {
case record.Samples: case record.Samples:
samples := samplesPool.Get().([]record.RefSample)[:0] samples := samplesPool.Get()[:0]
samples, err = dec.Samples(rec, samples) samples, err = dec.Samples(rec, samples)
if err != nil { if err != nil {
decodeErr = &wlog.CorruptionErr{ decodeErr = &wlog.CorruptionErr{
@ -727,7 +712,7 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
} }
decodedCh <- samples decodedCh <- samples
case record.MmapMarkers: case record.MmapMarkers:
markers := markersPool.Get().([]record.RefMmapMarker)[:0] markers := markersPool.Get()[:0]
markers, err = dec.MmapMarkers(rec, markers) markers, err = dec.MmapMarkers(rec, markers)
if err != nil { if err != nil {
decodeErr = &wlog.CorruptionErr{ decodeErr = &wlog.CorruptionErr{
@ -739,7 +724,7 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
} }
decodedCh <- markers decodedCh <- markers
case record.HistogramSamples: case record.HistogramSamples:
hists := histogramSamplesPool.Get().([]record.RefHistogramSample)[:0] hists := histogramSamplesPool.Get()[:0]
hists, err = dec.HistogramSamples(rec, hists) hists, err = dec.HistogramSamples(rec, hists)
if err != nil { if err != nil {
decodeErr = &wlog.CorruptionErr{ decodeErr = &wlog.CorruptionErr{
@ -751,7 +736,7 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
} }
decodedCh <- hists decodedCh <- hists
case record.FloatHistogramSamples: case record.FloatHistogramSamples:
hists := floatHistogramSamplesPool.Get().([]record.RefFloatHistogramSample)[:0] hists := floatHistogramSamplesPool.Get()[:0]
hists, err = dec.FloatHistogramSamples(rec, hists) hists, err = dec.FloatHistogramSamples(rec, hists)
if err != nil { if err != nil {
decodeErr = &wlog.CorruptionErr{ decodeErr = &wlog.CorruptionErr{
@ -802,7 +787,7 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
} }
samples = samples[m:] samples = samples[m:]
} }
samplesPool.Put(d) samplesPool.Put(v)
case []record.RefMmapMarker: case []record.RefMmapMarker:
markers := v markers := v
for _, rm := range markers { for _, rm := range markers {
@ -857,7 +842,7 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
} }
samples = samples[m:] samples = samples[m:]
} }
histogramSamplesPool.Put(v) //nolint:staticcheck histogramSamplesPool.Put(v)
case []record.RefFloatHistogramSample: case []record.RefFloatHistogramSample:
samples := v samples := v
// We split up the samples into chunks of 5000 samples or less. // We split up the samples into chunks of 5000 samples or less.
@ -889,7 +874,7 @@ func (h *Head) loadWBL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch
} }
samples = samples[m:] samples = samples[m:]
} }
floatHistogramSamplesPool.Put(v) //nolint:staticcheck floatHistogramSamplesPool.Put(v)
default: default:
panic(fmt.Errorf("unexpected decodedCh type: %T", d)) panic(fmt.Errorf("unexpected decodedCh type: %T", d))
} }