Skip to content

feat[scrape+head]: Implement EnableSTStorage for float scrapes #18100

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
bwplotka wants to merge 1 commit into from
Closed

feat[scrape+head]: Implement EnableSTStorage for float scrapes #18100

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion tsdb/db.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,6 @@ type Options struct {

// EnableSTStorage determines whether TSDB should write a Start Timestamp (ST)
// per sample to WAL.
// TODO(bwplotka): Implement this option as per PROM-60, currently it's noop.
EnableSTStorage bool

// EnableMetadataWALRecords represents 'metadata-wal-records' feature flag.
Expand Down
1 change: 0 additions & 1 deletion tsdb/head.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,6 @@ type HeadOptions struct {

// EnableSTStorage determines whether agent DB should write a Start Timestamp (ST)
// per sample to WAL.
// TODO(bwplotka): Implement this option as per PROM-60, currently it's noop.
EnableSTStorage bool
}

Expand Down
64 changes: 39 additions & 25 deletions tsdb/head_append.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1058,8 +1058,10 @@ func (a *headAppenderBase) log() error {
buf := a.head.getBytesBuffer()
defer func() { a.head.putBytesBuffer(buf) }()

var rec []byte
var enc record.Encoder
var (
rec []byte
enc = record.Encoder{EnableSTStorage: a.head.opts.EnableSTStorage}
)

if len(a.seriesRefs) > 0 {
rec = enc.Series(a.seriesRefs, buf)
Expand Down Expand Up @@ -1178,7 +1180,7 @@ type appenderCommitContext struct {
oooRecords [][]byte
oooCapMax int64
appendChunkOpts chunkOpts
enc record.Encoder
oooEnc record.Encoder

@bwplotka bwplotka Feb 18, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc - there was a bug

}

// commitExemplars adds all exemplars from the provided batch to the head's exemplar storage.
Expand Down Expand Up @@ -1228,31 +1230,31 @@ func (acc *appenderCommitContext) collectOOORecords(a *headAppenderBase) {
})
}
}
r := acc.enc.MmapMarkers(markers, a.head.getBytesBuffer())
r := acc.oooEnc.MmapMarkers(markers, a.head.getBytesBuffer())
acc.oooRecords = append(acc.oooRecords, r)
}

if len(acc.wblSamples) > 0 {
r := acc.enc.Samples(acc.wblSamples, a.head.getBytesBuffer())
r := acc.oooEnc.Samples(acc.wblSamples, a.head.getBytesBuffer())
acc.oooRecords = append(acc.oooRecords, r)
}
if len(acc.wblHistograms) > 0 {
r, customBucketsHistograms := acc.enc.HistogramSamples(acc.wblHistograms, a.head.getBytesBuffer())
r, customBucketsHistograms := acc.oooEnc.HistogramSamples(acc.wblHistograms, a.head.getBytesBuffer())
if len(r) > 0 {
acc.oooRecords = append(acc.oooRecords, r)
}
if len(customBucketsHistograms) > 0 {
r := acc.enc.CustomBucketsHistogramSamples(customBucketsHistograms, a.head.getBytesBuffer())
r := acc.oooEnc.CustomBucketsHistogramSamples(customBucketsHistograms, a.head.getBytesBuffer())
acc.oooRecords = append(acc.oooRecords, r)
}
}
if len(acc.wblFloatHistograms) > 0 {
r, customBucketsFloatHistograms := acc.enc.FloatHistogramSamples(acc.wblFloatHistograms, a.head.getBytesBuffer())
r, customBucketsFloatHistograms := acc.oooEnc.FloatHistogramSamples(acc.wblFloatHistograms, a.head.getBytesBuffer())
if len(r) > 0 {
acc.oooRecords = append(acc.oooRecords, r)
}
if len(customBucketsFloatHistograms) > 0 {
r := acc.enc.CustomBucketsFloatHistogramSamples(customBucketsFloatHistograms, a.head.getBytesBuffer())
r := acc.oooEnc.CustomBucketsFloatHistogramSamples(customBucketsFloatHistograms, a.head.getBytesBuffer())
acc.oooRecords = append(acc.oooRecords, r)
}
}
Expand Down Expand Up @@ -1431,7 +1433,7 @@ func (a *headAppenderBase) commitFloats(b *appendBatch, acc *appenderCommitConte
default:
newlyStale := !value.IsStaleNaN(series.lastValue) && value.IsStaleNaN(s.V)
staleToNonStale := value.IsStaleNaN(series.lastValue) && !value.IsStaleNaN(s.V)
ok, chunkCreated = series.append(s.T, s.V, a.appendID, acc.appendChunkOpts)
ok, chunkCreated = series.append(s.ST, s.T, s.V, a.appendID, acc.appendChunkOpts)
if ok {
if s.T < acc.inOrderMint {
acc.inOrderMint = s.T
Expand Down Expand Up @@ -1540,7 +1542,8 @@ func (a *headAppenderBase) commitHistograms(b *appendBatch, acc *appenderCommitC
newlyStale = newlyStale && !value.IsStaleNaN(series.lastHistogramValue.Sum)
staleToNonStale = value.IsStaleNaN(series.lastHistogramValue.Sum) && !value.IsStaleNaN(s.H.Sum)
}
ok, chunkCreated = series.appendHistogram(s.T, s.H, a.appendID, acc.appendChunkOpts)
// TODO(bwplotka): Add support for ST for Histograms.
ok, chunkCreated = series.appendHistogram(0, s.T, s.H, a.appendID, acc.appendChunkOpts)
if ok {
if s.T < acc.inOrderMint {
acc.inOrderMint = s.T
Expand Down Expand Up @@ -1649,7 +1652,8 @@ func (a *headAppenderBase) commitFloatHistograms(b *appendBatch, acc *appenderCo
newlyStale = newlyStale && !value.IsStaleNaN(series.lastFloatHistogramValue.Sum)
staleToNonStale = value.IsStaleNaN(series.lastFloatHistogramValue.Sum) && !value.IsStaleNaN(s.FH.Sum)
}
ok, chunkCreated = series.appendFloatHistogram(s.T, s.FH, a.appendID, acc.appendChunkOpts)
// TODO(bwplotka): Add support for ST for FloatHistograms.
ok, chunkCreated = series.appendFloatHistogram(0, s.T, s.FH, a.appendID, acc.appendChunkOpts)
if ok {
if s.T < acc.inOrderMint {
acc.inOrderMint = s.T
Expand Down Expand Up @@ -1741,9 +1745,10 @@ func (a *headAppenderBase) Commit() (err error) {
chunkDiskMapper: h.chunkDiskMapper,
chunkRange: h.chunkRange.Load(),
samplesPerChunk: h.opts.SamplesPerChunk,
enableSTStorage: h.opts.EnableSTStorage,
},
enc: record.Encoder{
EnableSTStorage: false,
oooEnc: record.Encoder{
EnableSTStorage: a.head.opts.EnableSharding,
},
}

Expand Down Expand Up @@ -1827,19 +1832,25 @@ type chunkOpts struct {
chunkDiskMapper *chunks.ChunkDiskMapper
chunkRange int64
samplesPerChunk int

enableSTStorage bool
}

// append adds the sample (t, v) to the series. The caller also has to provide
// the appendID for isolation. (The appendID can be zero, which results in no
// 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)
func (s *memSeries) append(st, t int64, v float64, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
enc := chunkenc.EncXOR
if o.enableSTStorage {

@bwplotka bwplotka Feb 19, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc

enc = chunkenc.EncXOROptST
}

c, sampleInOrder, chunkCreated := s.appendPreprocessor(t, enc, o)
if !sampleInOrder {
return sampleInOrder, chunkCreated
}
// TODO(krajorama): pass ST.
s.app.Append(0, t, v)
s.app.Append(st, t, v)

c.maxTime = t

Expand All @@ -1859,14 +1870,16 @@ func (s *memSeries) append(t int64, v float64, appendID uint64, o chunkOpts) (sa
// In case of recoding the existing chunk, a new chunk is allocated and the old chunk is dropped.
// To keep the meaning of prometheus_tsdb_head_chunks and prometheus_tsdb_head_chunks_created_total
// consistent, we return chunkCreated=false in this case.
func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
func (s *memSeries) appendHistogram(st, t int64, h *histogram.Histogram, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
// Head controls the execution of recoding, so that we own the proper
// chunk reference afterwards and mmap used up chunks.

// 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)
enc := chunkenc.EncHistogram
// TODO(bwplotka): Implement histogram ST encoding and switch on o.enableSTStorage.
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, enc, o)
if !sampleInOrder {
return sampleInOrder, chunkCreated
}
Expand All @@ -1882,7 +1895,7 @@ func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID ui
}

// TODO(krajorama): pass ST.
newChunk, recoded, s.app, _ = s.app.AppendHistogram(prevApp, 0, t, h, false) // false=request a new chunk if needed
newChunk, recoded, s.app, _ = s.app.AppendHistogram(prevApp, st, t, h, false) // false=request a new chunk if needed

s.lastHistogramValue = h
s.lastFloatHistogramValue = nil
Expand Down Expand Up @@ -1917,14 +1930,16 @@ func (s *memSeries) appendHistogram(t int64, h *histogram.Histogram, appendID ui
// In case of recoding the existing chunk, a new chunk is allocated and the old chunk is dropped.
// To keep the meaning of prometheus_tsdb_head_chunks and prometheus_tsdb_head_chunks_created_total
// consistent, we return chunkCreated=false in this case.
func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
func (s *memSeries) appendFloatHistogram(st, t int64, fh *histogram.FloatHistogram, appendID uint64, o chunkOpts) (sampleInOrder, chunkCreated bool) {
// Head controls the execution of recoding, so that we own the proper
// chunk reference afterwards and mmap used up chunks.

// 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)
enc := chunkenc.EncFloatHistogram
// TODO(bwplotka): Implement histogram ST encoding and switch on o.enableSTStorage.
c, sampleInOrder, chunkCreated := s.histogramsAppendPreprocessor(t, enc, o)
if !sampleInOrder {
return sampleInOrder, chunkCreated
}
Expand All @@ -1939,8 +1954,7 @@ func (s *memSeries) appendFloatHistogram(t int64, fh *histogram.FloatHistogram,
prevApp = nil
}

// TODO(krajorama): pass ST.
newChunk, recoded, s.app, _ = s.app.AppendFloatHistogram(prevApp, 0, t, fh, false) // False means request a new chunk if needed.
newChunk, recoded, s.app, _ = s.app.AppendFloatHistogram(prevApp, st, t, fh, false) // False means request a new chunk if needed.

s.lastHistogramValue = nil
s.lastFloatHistogramValue = fh
Expand Down
34 changes: 18 additions & 16 deletions tsdb/head_append_v2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,10 +148,10 @@ func (a *headAppenderV2) Append(ref storage.SeriesRef, ls labels.Labels, st, t i
switch {
case fh != nil:
isStale = value.IsStaleNaN(fh.Sum)
appErr = a.appendFloatHistogram(s, t, fh, opts.RejectOutOfOrder)
appErr = a.appendFloatHistogram(s, st, t, fh, opts.RejectOutOfOrder)
case h != nil:
isStale = value.IsStaleNaN(h.Sum)
appErr = a.appendHistogram(s, t, h, opts.RejectOutOfOrder)
appErr = a.appendHistogram(s, st, t, h, opts.RejectOutOfOrder)
default:
isStale = value.IsStaleNaN(v)
if isStale {
Expand All @@ -177,7 +177,7 @@ func (a *headAppenderV2) Append(ref storage.SeriesRef, ls labels.Labels, st, t i
// we do not need to check for the difference between "unknown
// series" and "known series with stNone".
}
appErr = a.appendFloat(s, t, v, opts.RejectOutOfOrder)
appErr = a.appendFloat(s, st, t, v, opts.RejectOutOfOrder)
}
// Handle append error, if any.
if appErr != nil {
Expand Down Expand Up @@ -218,7 +218,7 @@ func (a *headAppenderV2) Append(ref storage.SeriesRef, ls labels.Labels, st, t i
return storage.SeriesRef(s.ref), partialErr
}

func (a *headAppenderV2) appendFloat(s *memSeries, t int64, v float64, fastRejectOOO bool) error {
func (a *headAppenderV2) appendFloat(s *memSeries, st, t int64, v float64, fastRejectOOO bool) error {
s.Lock()
// TODO(codesome): If we definitely know at this point that the sample is ooo, then optimise
// to skip that sample from the WAL and write only in the WBL.
Expand All @@ -239,12 +239,12 @@ func (a *headAppenderV2) appendFloat(s *memSeries, t int64, v float64, fastRejec
}

b := a.getCurrentBatch(stFloat, s.ref)
b.floats = append(b.floats, record.RefSample{Ref: s.ref, T: t, V: v})
b.floats = append(b.floats, record.RefSample{Ref: s.ref, ST: st, T: t, V: v})
b.floatSeries = append(b.floatSeries, s)
return nil
}

func (a *headAppenderV2) appendHistogram(s *memSeries, t int64, h *histogram.Histogram, fastRejectOOO bool) error {
func (a *headAppenderV2) appendHistogram(s *memSeries, _, t int64, h *histogram.Histogram, fastRejectOOO bool) error {
s.Lock()
// TODO(codesome): If we definitely know at this point that the sample is ooo, then optimise
// to skip that sample from the WAL and write only in the WBL.
Expand All @@ -263,17 +263,18 @@ func (a *headAppenderV2) appendHistogram(s *memSeries, t int64, h *histogram.His
if err != nil {
return err
}
st := stHistogram
sTyp := stHistogram
if h.UsesCustomBuckets() {
st = stCustomBucketHistogram
sTyp = stCustomBucketHistogram
}
b := a.getCurrentBatch(st, s.ref)
b := a.getCurrentBatch(sTyp, s.ref)
// TODO(bwplotka): Add ST support for RefHistogramSample.
b.histograms = append(b.histograms, record.RefHistogramSample{Ref: s.ref, T: t, H: h})
b.histogramSeries = append(b.histogramSeries, s)
return nil
}

func (a *headAppenderV2) appendFloatHistogram(s *memSeries, t int64, fh *histogram.FloatHistogram, fastRejectOOO bool) error {
func (a *headAppenderV2) appendFloatHistogram(s *memSeries, _, t int64, fh *histogram.FloatHistogram, fastRejectOOO bool) error {
s.Lock()
// TODO(codesome): If we definitely know at this point that the sample is ooo, then optimise
// to skip that sample from the WAL and write only in the WBL.
Expand All @@ -292,11 +293,12 @@ func (a *headAppenderV2) appendFloatHistogram(s *memSeries, t int64, fh *histogr
if err != nil {
return err
}
st := stFloatHistogram
sTyp := stFloatHistogram
if fh.UsesCustomBuckets() {
st = stCustomBucketFloatHistogram
sTyp = stCustomBucketFloatHistogram
}
b := a.getCurrentBatch(st, s.ref)
b := a.getCurrentBatch(sTyp, s.ref)
// TODO(bwplotka): Add ST support for RefFloatHistogramSample.
b.floatHistograms = append(b.floatHistograms, record.RefFloatHistogramSample{Ref: s.ref, T: t, FH: fh})
b.floatHistogramSeries = append(b.floatHistogramSeries, s)
return nil
Expand Down Expand Up @@ -354,7 +356,7 @@ func (a *headAppenderV2) bestEffortAppendSTZeroSample(s *memSeries, ls labels.La
ZeroThreshold: fh.ZeroThreshold,
CustomValues: fh.CustomValues,
}
err = a.appendFloatHistogram(s, st, zeroFloatHistogram, true)
err = a.appendFloatHistogram(s, 0, st, zeroFloatHistogram, true)
case h != nil:
zeroHistogram := &histogram.Histogram{
// The STZeroSample represents a counter reset by definition.
Expand All @@ -364,9 +366,9 @@ func (a *headAppenderV2) bestEffortAppendSTZeroSample(s *memSeries, ls labels.La
ZeroThreshold: h.ZeroThreshold,
CustomValues: h.CustomValues,
}
err = a.appendHistogram(s, st, zeroHistogram, true)
err = a.appendHistogram(s, 0, st, zeroHistogram, true)
default:
err = a.appendFloat(s, st, 0, true)
err = a.appendFloat(s, 0, st, 0, true)
}

if err != nil {
Expand Down
Loading
Loading