mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-09-19 23:37:31 -07:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

TSDB: NewHeadAndOOOChunkReader takes headChunkReader

Browse source 
So we can pass nil and have it read just OOO chunks.

Signed-off-by: Bryan Boreham <bjboreham@gmail.com>
This commit is contained in:
Bryan Boreham 2024-07-15 20:07:12 +01:00
parent e04d137649
commit 6529d6336c
1 changed files with 25 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
tsdb/ooo_head_read.go
View file

@ -514,13 +514,18 @@ type HeadAndOOOQuerier struct {
} }
func NewHeadAndOOOQuerier(mint, maxt int64, head *Head, oooIsoState *oooIsolationState, querier storage.Querier) storage.Querier { func NewHeadAndOOOQuerier(mint, maxt int64, head *Head, oooIsoState *oooIsolationState, querier storage.Querier) storage.Querier {
isoState := head.iso.State(mint, maxt) cr := &headChunkReader{
head: head,
mint: mint,
maxt: maxt,
isoState: head.iso.State(mint, maxt),
}
return &HeadAndOOOQuerier{ return &HeadAndOOOQuerier{
mint: mint, mint: mint,
maxt: maxt, maxt: maxt,
head: head, head: head,
index: NewHeadAndOOOIndexReader(head, mint, maxt, oooIsoState.minRef), index: NewHeadAndOOOIndexReader(head, mint, maxt, oooIsoState.minRef),
chunkr: NewHeadAndOOOChunkReader(head, mint, maxt, isoState, oooIsoState, 0), chunkr: NewHeadAndOOOChunkReader(head, mint, maxt, cr, oooIsoState, 0),
querier: querier, querier: querier,
} }
} }
@ -552,13 +557,18 @@ type HeadAndOOOChunkQuerier struct {
} }
func NewHeadAndOOOChunkQuerier(mint, maxt int64, head *Head, oooIsoState *oooIsolationState, querier storage.ChunkQuerier) storage.ChunkQuerier { func NewHeadAndOOOChunkQuerier(mint, maxt int64, head *Head, oooIsoState *oooIsolationState, querier storage.ChunkQuerier) storage.ChunkQuerier {
isoState := head.iso.State(mint, maxt) cr := &headChunkReader{
head: head,
mint: mint,
maxt: maxt,
isoState: head.iso.State(mint, maxt),
}
return &HeadAndOOOChunkQuerier{ return &HeadAndOOOChunkQuerier{
mint: mint, mint: mint,
maxt: maxt, maxt: maxt,
head: head, head: head,
index: NewHeadAndOOOIndexReader(head, mint, maxt, oooIsoState.minRef), index: NewHeadAndOOOIndexReader(head, mint, maxt, oooIsoState.minRef),
chunkr: NewHeadAndOOOChunkReader(head, mint, maxt, isoState, oooIsoState, 0), chunkr: NewHeadAndOOOChunkReader(head, mint, maxt, cr, oooIsoState, 0),
querier: querier, querier: querier,
} }
} }
@ -618,19 +628,19 @@ func (oh *HeadAndOOOIndexReader) Series(ref storage.SeriesRef, builder *labels.S
} }
type HeadAndOOOChunkReader struct { type HeadAndOOOChunkReader struct {
cr headChunkReader head *Head
mint, maxt int64
cr *headChunkReader // If nil, only read OOO chunks.
maxMmapRef chunks.ChunkDiskMapperRef maxMmapRef chunks.ChunkDiskMapperRef
oooIsoState *oooIsolationState oooIsoState *oooIsolationState
} }
func NewHeadAndOOOChunkReader(head *Head, mint, maxt int64, isoState *isolationState, oooIsoState *oooIsolationState, maxMmapRef chunks.ChunkDiskMapperRef) *HeadAndOOOChunkReader { func NewHeadAndOOOChunkReader(head *Head, mint, maxt int64, cr *headChunkReader, oooIsoState *oooIsolationState, maxMmapRef chunks.ChunkDiskMapperRef) *HeadAndOOOChunkReader {
return &HeadAndOOOChunkReader{ return &HeadAndOOOChunkReader{
cr: headChunkReader{ head: head,
head: head, mint: mint,
mint: mint, maxt: maxt,
maxt: maxt, cr: cr,
isoState: isoState,
},
maxMmapRef: maxMmapRef, maxMmapRef: maxMmapRef,
oooIsoState: oooIsoState, oooIsoState: oooIsoState,
} }
@ -642,14 +652,14 @@ func (cr *HeadAndOOOChunkReader) ChunkOrIterable(meta chunks.Meta) (chunkenc.Chu
return cr.cr.ChunkOrIterable(meta) return cr.cr.ChunkOrIterable(meta)
} }
s := cr.cr.head.series.getByID(sid) s := cr.head.series.getByID(sid)
// This means that the series has been garbage collected. // This means that the series has been garbage collected.
if s == nil { if s == nil {
return nil, nil, storage.ErrNotFound return nil, nil, storage.ErrNotFound
} }
s.Lock() s.Lock()
mc, err := s.oooMergedChunks(meta, cr.cr.head.chunkDiskMapper, &cr.cr, cr.cr.mint, cr.cr.maxt, cr.maxMmapRef) mc, err := s.oooMergedChunks(meta, cr.head.chunkDiskMapper, cr.cr, cr.mint, cr.maxt, cr.maxMmapRef)
s.Unlock() s.Unlock()
return nil, mc, err return nil, mc, err
@ -666,7 +676,7 @@ func (cr *HeadAndOOOChunkReader) ChunkOrIterableWithCopy(meta chunks.Meta) (chun
} }
func (cr *HeadAndOOOChunkReader) Close() error { func (cr *HeadAndOOOChunkReader) Close() error {
if cr.cr.isoState != nil { if cr.cr != nil && cr.cr.isoState != nil {
cr.cr.isoState.Close() cr.cr.isoState.Close()
} }
if cr.oooIsoState != nil { if cr.oooIsoState != nil {