mirrors/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2026-05-13 08:36:38 +02:00
Code Issues Projects Releases Wiki Activity

M-map a series as soon as it goes stale

Browse Source 
Signed-off-by: Ganesh Vernekar <ganesh.vernekar@reddit.com>
This commit is contained in:
Ganesh Vernekar 2025-07-01 13:47:16 -07:00
parent d902abc50d
commit 5c8edf69bd
1 changed files with 19 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
tsdb/head_append.go
View File

@ -1552,7 +1552,7 @@ type chunkOpts struct {
// isolation for this append.)
// Series lock must be held when calling.
func (s *memSeries) append(t int64, v float64, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
c, sampleInOrder, chunkCreated := s.appendPreprocessor(t, chunkenc.EncXOR, o)
c, sampleInOrder, chunkCreated := s.appendPreprocessor(t, value.IsStaleNaN(v), chunkenc.EncXOR, o)
if !sampleInOrder {
return sampleInOrder, chunkCreated
}
@ -1583,7 +1583,7 @@ func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID ui
// Ignoring ok is ok, since we don't want to compare to the wrong previous appender anyway.
prevApp, _ := s.app.(*chunkenc.HistogramAppender)
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, chunkenc.EncHistogram, o)
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, value.IsStaleNaN(h.Sum), chunkenc.EncHistogram, o)
if !sampleInOrder {
return sampleInOrder, chunkCreated
}
@ -1640,7 +1640,7 @@ func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram,
// Ignoring ok is ok, since we don't want to compare to the wrong previous appender anyway.
prevApp, _ := s.app.(*chunkenc.FloatHistogramAppender)
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, chunkenc.EncFloatHistogram, o)
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, value.IsStaleNaN(fh.Sum), chunkenc.EncFloatHistogram, o)
if !sampleInOrder {
return sampleInOrder, chunkCreated
}
@ -1689,7 +1689,7 @@ func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram,
// number of samples they contain with a soft cap in bytes.
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
// This should be called only when appending data.
func (s *memSeries) appendPreprocessor(t int64, e chunkenc.Encoding, o chunkOpts) (c *memChunk, sampleInOrder, chunkCreated bool) {
func (s *memSeries) appendPreprocessor(t int64, isStale bool, e chunkenc.Encoding, o chunkOpts) (c *memChunk, sampleInOrder, chunkCreated bool) {
// We target chunkenc.MaxBytesPerXORChunk as a hard for the size of an XOR chunk. We must determine whether to cut
// a new head chunk without knowing the size of the next sample, however, so we assume the next sample will be a
// maximally-sized sample (19 bytes).
@ -1725,6 +1725,13 @@ func (s *memSeries) appendPreprocessor(t int64, e chunkenc.Encoding, o chunkOpts
chunkCreated = true
}
if isStale && !chunkCreated {
// This series has gone stale, we can save some memory and just flush its
// data to the disk.
c = s.cutNewHeadChunk(t, e, o.chunkRange)
chunkCreated = true
}
numSamples := c.chunk.NumSamples()
if numSamples == 0 {
// It could be the new chunk created after reading the chunk snapshot,
@ -1757,7 +1764,7 @@ func (s *memSeries) appendPreprocessor(t int64, e chunkenc.Encoding, o chunkOpts
// cut based on their size in bytes.
// It is unsafe to call this concurrently with s.iterator(...) without holding the series lock.
// This should be called only when appending data.
func (s *memSeries) histogramsAppendPreprocessor(t int64, e chunkenc.Encoding, o chunkOpts) (c *memChunk, sampleInOrder, chunkCreated bool) {
func (s *memSeries) histogramsAppendPreprocessor(t int64, isStale bool, e chunkenc.Encoding, o chunkOpts) (c *memChunk, sampleInOrder, chunkCreated bool) {
c = s.headChunks
if c == nil {
@ -1782,6 +1789,13 @@ func (s *memSeries) histogramsAppendPreprocessor(t int64, e chunkenc.Encoding, o
chunkCreated = true
}
if isStale && !chunkCreated {
// This series has gone stale, we can save some memory and just flush its
// data to the disk.
c = s.cutNewHeadChunk(t, e, o.chunkRange)
chunkCreated = true
}
numSamples := c.chunk.NumSamples()
targetBytes := chunkenc.TargetBytesPerHistogramChunk
numBytes := len(c.chunk.Bytes())