diff --git a/storage/remote/queue_manager.go b/storage/remote/queue_manager.go index 98ad7d72860..6f0a42fcac1 100644 --- a/storage/remote/queue_manager.go +++ b/storage/remote/queue_manager.go @@ -787,6 +787,60 @@ outer: return true } +func (t *QueueManager) AppendSamplesV2(samples []record.RefSampleV2) bool { + currentTime := time.Now() +outer: + for _, s := range samples { + if isSampleOld(currentTime, time.Duration(t.cfg.SampleAgeLimit), s.T) { + t.metrics.droppedSamplesTotal.WithLabelValues(reasonTooOld).Inc() + continue + } + t.seriesMtx.Lock() + lbls, ok := t.seriesLabels[s.Ref] + if !ok { + t.dataDropped.incr(1) + if _, ok := t.droppedSeries[s.Ref]; !ok { + t.logger.Info("Dropped sample for series that was not explicitly dropped via relabelling", "ref", s.Ref) + t.metrics.droppedSamplesTotal.WithLabelValues(reasonUnintentionalDroppedSeries).Inc() + } else { + t.metrics.droppedSamplesTotal.WithLabelValues(reasonDroppedSeries).Inc() + } + t.seriesMtx.Unlock() + continue + } + meta := t.seriesMetadata[s.Ref] + t.seriesMtx.Unlock() + + backoff := model.Duration(5 * time.Millisecond) + for { + select { + case <-t.quit: + return false + default: + } + if t.shards.enqueue(s.Ref, timeSeries{ + seriesLabels: lbls, + metadata: meta, + startTimestamp: s.ST, + timestamp: s.T, + value: s.V, + exemplars: s.Exemplars, + sType: tSample, + }) { + continue outer + } + + t.metrics.enqueueRetriesTotal.Inc() + time.Sleep(time.Duration(backoff)) + backoff *= 2 + if backoff > t.cfg.MaxBackoff { + backoff = t.cfg.MaxBackoff + } + } + } + return true +} + func (t *QueueManager) AppendExemplars(exemplars []record.RefExemplar) bool { if !t.sendExemplars { return true @@ -906,6 +960,69 @@ outer: return true } +func (t *QueueManager) AppendHistogramsV2(histograms []record.RefHistogramSampleV2) bool { + if !t.sendNativeHistograms { + return true + } + currentTime := time.Now() +outer: + for _, h := range histograms { + if isSampleOld(currentTime, time.Duration(t.cfg.SampleAgeLimit), h.T) { + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonTooOld).Inc() + continue + } + if t.protoMsg == remoteapi.WriteV1MessageType && h.H != nil && h.H.Schema == histogram.CustomBucketsSchema { + // We cannot send native histograms with custom buckets (NHCB) via remote write v1. + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonNHCBNotSupported).Inc() + t.logger.Warn("Dropped native histogram with custom buckets (NHCB) as remote write v1 does not support itB", "ref", h.Ref) + continue + } + t.seriesMtx.Lock() + lbls, ok := t.seriesLabels[h.Ref] + if !ok { + t.dataDropped.incr(1) + if _, ok := t.droppedSeries[h.Ref]; !ok { + t.logger.Info("Dropped histogram for series that was not explicitly dropped via relabelling", "ref", h.Ref) + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonUnintentionalDroppedSeries).Inc() + } else { + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonDroppedSeries).Inc() + } + t.seriesMtx.Unlock() + continue + } + meta := t.seriesMetadata[h.Ref] + t.seriesMtx.Unlock() + + backoff := model.Duration(5 * time.Millisecond) + for { + select { + case <-t.quit: + return false + default: + } + if t.shards.enqueue(h.Ref, timeSeries{ + seriesLabels: lbls, + metadata: meta, + startTimestamp: h.ST, + timestamp: h.T, + histogram: h.H, + exemplars: h.Exemplars, + sType: tHistogram, + }) { + continue outer + } + + t.metrics.enqueueRetriesTotal.Inc() + time.Sleep(time.Duration(backoff)) + backoff *= 2 + if backoff > t.cfg.MaxBackoff { + backoff = t.cfg.MaxBackoff + } + } + } + return true +} + func (t *QueueManager) AppendFloatHistograms(floatHistograms []record.RefFloatHistogramSample) bool { if !t.sendNativeHistograms { return true @@ -968,6 +1085,69 @@ outer: return true } +func (t *QueueManager) AppendFloatHistogramsV2(floatHistograms []record.RefFloatHistogramSampleV2) bool { + if !t.sendNativeHistograms { + return true + } + currentTime := time.Now() +outer: + for _, h := range floatHistograms { + if isSampleOld(currentTime, time.Duration(t.cfg.SampleAgeLimit), h.T) { + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonTooOld).Inc() + continue + } + if t.protoMsg == remoteapi.WriteV1MessageType && h.FH != nil && h.FH.Schema == histogram.CustomBucketsSchema { + // We cannot send native histograms with custom buckets (NHCB) via remote write v1. + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonNHCBNotSupported).Inc() + t.logger.Warn("Dropped float native histogram with custom buckets (NHCB) as remote write v1 does not support itB", "ref", h.Ref) + continue + } + t.seriesMtx.Lock() + lbls, ok := t.seriesLabels[h.Ref] + if !ok { + t.dataDropped.incr(1) + if _, ok := t.droppedSeries[h.Ref]; !ok { + t.logger.Info("Dropped histogram for series that was not explicitly dropped via relabelling", "ref", h.Ref) + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonUnintentionalDroppedSeries).Inc() + } else { + t.metrics.droppedHistogramsTotal.WithLabelValues(reasonDroppedSeries).Inc() + } + t.seriesMtx.Unlock() + continue + } + meta := t.seriesMetadata[h.Ref] + t.seriesMtx.Unlock() + + backoff := model.Duration(5 * time.Millisecond) + for { + select { + case <-t.quit: + return false + default: + } + if t.shards.enqueue(h.Ref, timeSeries{ + seriesLabels: lbls, + metadata: meta, + startTimestamp: h.ST, + timestamp: h.T, + floatHistogram: h.FH, + exemplars: h.Exemplars, + sType: tFloatHistogram, + }) { + continue outer + } + + t.metrics.enqueueRetriesTotal.Inc() + time.Sleep(time.Duration(backoff)) + backoff *= 2 + if backoff > t.cfg.MaxBackoff { + backoff = t.cfg.MaxBackoff + } + } + } + return true +} + // Start the queue manager sending samples to the remote storage. // Does not block. func (t *QueueManager) Start() { @@ -1410,6 +1590,7 @@ type timeSeries struct { metadata *metadata.Metadata startTimestamp, timestamp int64 exemplarLabels labels.Labels + exemplars []record.RefExemplar // The type of series: sample, exemplar, or histogram. sType seriesType } @@ -1679,6 +1860,16 @@ func populateTimeSeries(batch []timeSeries, pendingData []prompb.TimeSeries, sen Timestamp: d.timestamp, }) nPendingSamples++ + if sendExemplars && len(d.exemplars) > 0 { + for _, ex := range d.exemplars { + pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.Exemplar{ + Labels: prompb.FromLabels(ex.Labels, nil), + Value: ex.V, + Timestamp: ex.T, + }) + nPendingExemplars++ + } + } case tExemplar: pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.Exemplar{ Labels: prompb.FromLabels(d.exemplarLabels, nil), @@ -1689,9 +1880,29 @@ func populateTimeSeries(batch []timeSeries, pendingData []prompb.TimeSeries, sen case tHistogram: pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, prompb.FromIntHistogram(d.timestamp, d.histogram)) nPendingHistograms++ + if sendExemplars && len(d.exemplars) > 0 { + for _, ex := range d.exemplars { + pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.Exemplar{ + Labels: prompb.FromLabels(ex.Labels, nil), + Value: ex.V, + Timestamp: ex.T, + }) + nPendingExemplars++ + } + } case tFloatHistogram: pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, prompb.FromFloatHistogram(d.timestamp, d.floatHistogram)) nPendingHistograms++ + if sendExemplars && len(d.exemplars) > 0 { + for _, ex := range d.exemplars { + pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.Exemplar{ + Labels: prompb.FromLabels(ex.Labels, nil), + Value: ex.V, + Timestamp: ex.T, + }) + nPendingExemplars++ + } + } } } return nPendingSamples, nPendingExemplars, nPendingHistograms @@ -2017,6 +2228,16 @@ func populateV2TimeSeries(symbolTable *writev2.SymbolsTable, batch []timeSeries, StartTimestamp: d.startTimestamp, }) nPendingSamples++ + if sendExemplars && len(d.exemplars) > 0 { + for _, ex := range d.exemplars { + pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, writev2.Exemplar{ + LabelsRefs: symbolTable.SymbolizeLabels(ex.Labels, nil), + Value: ex.V, + Timestamp: ex.T, + }) + nPendingExemplars++ + } + } case tExemplar: pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, writev2.Exemplar{ LabelsRefs: symbolTable.SymbolizeLabels(d.exemplarLabels, nil), // TODO: optimize, reuse slice @@ -2027,9 +2248,29 @@ func populateV2TimeSeries(symbolTable *writev2.SymbolsTable, batch []timeSeries, case tHistogram: pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, writev2.FromIntHistogram(d.startTimestamp, d.timestamp, d.histogram)) nPendingHistograms++ + if sendExemplars && len(d.exemplars) > 0 { + for _, ex := range d.exemplars { + pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, writev2.Exemplar{ + LabelsRefs: symbolTable.SymbolizeLabels(ex.Labels, nil), + Value: ex.V, + Timestamp: ex.T, + }) + nPendingExemplars++ + } + } case tFloatHistogram: pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, writev2.FromFloatHistogram(d.startTimestamp, d.timestamp, d.floatHistogram)) nPendingHistograms++ + if sendExemplars && len(d.exemplars) > 0 { + for _, ex := range d.exemplars { + pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, writev2.Exemplar{ + LabelsRefs: symbolTable.SymbolizeLabels(ex.Labels, nil), + Value: ex.V, + Timestamp: ex.T, + }) + nPendingExemplars++ + } + } case tMetadata: nUnexpectedMetadata++ } diff --git a/storage/remote/queue_manager_test.go b/storage/remote/queue_manager_test.go index 90236200ff7..69171fb65ea 100644 --- a/storage/remote/queue_manager_test.go +++ b/storage/remote/queue_manager_test.go @@ -2808,3 +2808,97 @@ func TestAppendHistogramsWithStartTimestamp(t *testing.T) { c.waitForExpectedData(t, 30*time.Second) } + +func TestQueueManager_PRW2_ExemplarAttachment(t *testing.T) { + lbls := labels.FromStrings("__name__", "http_requests_total", "job", "api") + ex1 := record.RefExemplar{ + Ref: 1, + T: 1000, + V: 42.0, + Labels: labels.FromStrings("trace_id", "abc-123"), + } + ex2 := record.RefExemplar{ + Ref: 1, + T: 2000, + V: 43.0, + Labels: labels.FromStrings("trace_id", "def-456"), + } + + batch := []timeSeries{ + { + seriesLabels: lbls, + value: 42.0, + timestamp: 1000, + startTimestamp: 500, + exemplars: []record.RefExemplar{ex1}, + sType: tSample, + }, + { + seriesLabels: lbls, + value: 43.0, + timestamp: 2000, + startTimestamp: 500, + exemplars: []record.RefExemplar{ex2}, + sType: tSample, + }, + } + + symbolTable := writev2.NewSymbolTable() + pendingData := make([]writev2.TimeSeries, len(batch)) + + nSamples, nExemplars, nHistograms, _, _ := populateV2TimeSeries(&symbolTable, batch, pendingData, true, true, false) + + require.Equal(t, 2, nSamples) + require.Equal(t, 2, nExemplars) + require.Equal(t, 0, nHistograms) + + // Verify each time series has BOTH sample AND attached exemplar + for i, ts := range pendingData { + require.Len(t, ts.Samples, 1, "expected 1 sample on series %d", i) + require.Len(t, ts.Exemplars, 1, "expected 1 exemplar attached to series %d", i) + require.Equal(t, batch[i].value, ts.Samples[0].Value) + require.Equal(t, batch[i].timestamp, ts.Samples[0].Timestamp) + require.Equal(t, batch[i].exemplars[0].V, ts.Exemplars[0].Value) + } + + // Verify invariant: NO standalone exemplar series (series with 0 samples/histograms and > 0 exemplars) + for i, ts := range pendingData { + if len(ts.Exemplars) > 0 { + require.True(t, len(ts.Samples) > 0 || len(ts.Histograms) > 0, "series %d has exemplars but 0 samples/histograms", i) + } + } +} + +func TestQueueManager_PRW1_ExemplarAttachment(t *testing.T) { + lbls := labels.FromStrings("__name__", "http_requests_total", "job", "api") + ex1 := record.RefExemplar{ + Ref: 1, + T: 1000, + V: 42.0, + Labels: labels.FromStrings("trace_id", "abc-123"), + } + + batch := []timeSeries{ + { + seriesLabels: lbls, + value: 42.0, + timestamp: 1000, + startTimestamp: 500, + exemplars: []record.RefExemplar{ex1}, + sType: tSample, + }, + } + + pendingData := make([]prompb.TimeSeries, len(batch)) + nSamples, nExemplars, nHistograms := populateTimeSeries(batch, pendingData, true, true) + + require.Equal(t, 1, nSamples) + require.Equal(t, 1, nExemplars) + require.Equal(t, 0, nHistograms) + + require.Len(t, pendingData[0].Samples, 1) + require.Len(t, pendingData[0].Exemplars, 1) + require.Equal(t, 42.0, pendingData[0].Samples[0].Value) + require.Equal(t, 42.0, pendingData[0].Exemplars[0].Value) + require.Equal(t, int64(1000), pendingData[0].Exemplars[0].Timestamp) +} diff --git a/tsdb/head_append.go b/tsdb/head_append.go index a6dcd0d274b..a30aa0fbdf7 100644 --- a/tsdb/head_append.go +++ b/tsdb/head_append.go @@ -374,29 +374,35 @@ const ( // because it is unclear if it is needed at all. (Maybe we will remove metadata // records altogether, see issue #15911.) type appendBatch struct { - floats []record.RefSample // New float samples held by this appender. - floatSeries []*memSeries // Float series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once). - histograms []record.RefHistogramSample // New histogram samples held by this appender. - histogramSeries []*memSeries // HistogramSamples series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once). - floatHistograms []record.RefFloatHistogramSample // New float histogram samples held by this appender. - floatHistogramSeries []*memSeries // FloatHistogramSamples series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once). - metadata []record.RefMetadata // New metadata held by this appender. - metadataSeries []*memSeries // Series corresponding to the metadata held by this appender. - exemplars []exemplarWithSeriesRef // New exemplars held by this appender. + floats []record.RefSample // New float samples held by this appender. + floatsV2 []record.RefSampleV2 // New V2 float samples with attached exemplars. + floatSeries []*memSeries // Float series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once). + histograms []record.RefHistogramSample // New histogram samples held by this appender. + histogramsV2 []record.RefHistogramSampleV2 // New V2 histogram samples with attached exemplars. + histogramSeries []*memSeries // HistogramSamples series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once). + floatHistograms []record.RefFloatHistogramSample // New float histogram samples held by this appender. + floatHistogramsV2 []record.RefFloatHistogramSampleV2 // New V2 float histogram samples with attached exemplars. + floatHistogramSeries []*memSeries // FloatHistogramSamples series corresponding to the samples held by this appender (using corresponding slice indices - same series may appear more than once). + metadata []record.RefMetadata // New metadata held by this appender. + metadataSeries []*memSeries // Series corresponding to the metadata held by this appender. + exemplars []exemplarWithSeriesRef // New exemplars held by this appender. } // close returns all the slices to the pools in Head and nil's them. func (b *appendBatch) close(h *Head) { h.putFloatBuffer(b.floats) b.floats = nil + b.floatsV2 = nil h.putSeriesBuffer(b.floatSeries) b.floatSeries = nil h.putHistogramBuffer(b.histograms) b.histograms = nil + b.histogramsV2 = nil h.putSeriesBuffer(b.histogramSeries) b.histogramSeries = nil h.putFloatHistogramBuffer(b.floatHistograms) b.floatHistograms = nil + b.floatHistogramsV2 = nil h.putSeriesBuffer(b.floatHistogramSeries) b.floatHistogramSeries = nil h.putMetadataBuffer(b.metadata) @@ -1137,7 +1143,22 @@ func (a *headAppenderBase) log() error { } // It's important to do (float) Samples before histogram samples // to end up with the correct order. - if len(b.floats) > 0 { + if len(b.floatsV2) > 0 { + if a.storeST || hasExemplarsFloats(b.floatsV2) { + rec = enc.SamplesV2(b.floatsV2, buf) + } else { + v1Samples := make([]record.RefSample, len(b.floatsV2)) + for i, s := range b.floatsV2 { + v1Samples[i] = record.RefSample{Ref: s.Ref, ST: s.ST, T: s.T, V: s.V} + } + rec = enc.Samples(v1Samples, buf) + } + buf = rec[:0] + + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log samples: %w", err) + } + } else if len(b.floats) > 0 { rec = enc.Samples(b.floats, buf) buf = rec[:0] @@ -1145,7 +1166,37 @@ func (a *headAppenderBase) log() error { return fmt.Errorf("log samples: %w", err) } } - if len(b.histograms) > 0 { + if len(b.histogramsV2) > 0 { + if a.storeST || hasExemplarsHistograms(b.histogramsV2) { + rec = enc.HistogramSamplesV2(b.histogramsV2, buf) + buf = rec[:0] + if len(rec) > 0 { + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log histograms v2: %w", err) + } + } + } else { + v1Histograms := make([]record.RefHistogramSample, len(b.histogramsV2)) + for i, h := range b.histogramsV2 { + v1Histograms[i] = record.RefHistogramSample{Ref: h.Ref, ST: h.ST, T: h.T, H: h.H} + } + var customBucketsHistograms []record.RefHistogramSample + rec, customBucketsHistograms = enc.HistogramSamples(v1Histograms, buf) + buf = rec[:0] + if len(rec) > 0 { + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log histograms: %w", err) + } + } + + if len(customBucketsHistograms) > 0 { + rec = enc.CustomBucketsHistogramSamples(customBucketsHistograms, buf) + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log custom buckets histograms: %w", err) + } + } + } + } else if len(b.histograms) > 0 { var customBucketsHistograms []record.RefHistogramSample rec, customBucketsHistograms = enc.HistogramSamples(b.histograms, buf) buf = rec[:0] @@ -1162,7 +1213,37 @@ func (a *headAppenderBase) log() error { } } } - if len(b.floatHistograms) > 0 { + if len(b.floatHistogramsV2) > 0 { + if a.storeST || hasExemplarsFloatHistograms(b.floatHistogramsV2) { + rec = enc.FloatHistogramSamplesV2(b.floatHistogramsV2, buf) + buf = rec[:0] + if len(rec) > 0 { + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log float histograms v2: %w", err) + } + } + } else { + v1FloatHistograms := make([]record.RefFloatHistogramSample, len(b.floatHistogramsV2)) + for i, fh := range b.floatHistogramsV2 { + v1FloatHistograms[i] = record.RefFloatHistogramSample{Ref: fh.Ref, ST: fh.ST, T: fh.T, FH: fh.FH} + } + var customBucketsFloatHistograms []record.RefFloatHistogramSample + rec, customBucketsFloatHistograms = enc.FloatHistogramSamples(v1FloatHistograms, buf) + buf = rec[:0] + if len(rec) > 0 { + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log float histograms: %w", err) + } + } + + if len(customBucketsFloatHistograms) > 0 { + rec = enc.CustomBucketsFloatHistogramSamples(customBucketsFloatHistograms, buf) + if err := a.head.wal.Log(rec); err != nil { + return fmt.Errorf("log custom buckets float histograms: %w", err) + } + } + } + } else if len(b.floatHistograms) > 0 { var customBucketsFloatHistograms []record.RefFloatHistogramSample rec, customBucketsFloatHistograms = enc.FloatHistogramSamples(b.floatHistograms, buf) buf = rec[:0] @@ -1183,7 +1264,8 @@ func (a *headAppenderBase) log() error { // otherwise it might happen that we send the exemplars in a remote write // batch before the samples, which in turn means the exemplar is rejected // for missing series, since series are created due to samples. - if len(b.exemplars) > 0 { + // For compound V2 records, exemplars are already coupled in the sample records. + if len(b.floatsV2) == 0 && len(b.histogramsV2) == 0 && len(b.floatHistogramsV2) == 0 && len(b.exemplars) > 0 { rec = enc.Exemplars(exemplarsForEncoding(b.exemplars), buf) buf = rec[:0] @@ -1195,6 +1277,33 @@ func (a *headAppenderBase) log() error { return nil } +func hasExemplarsFloats(samples []record.RefSampleV2) bool { + for _, s := range samples { + if len(s.Exemplars) > 0 { + return true + } + } + return false +} + +func hasExemplarsHistograms(samples []record.RefHistogramSampleV2) bool { + for _, s := range samples { + if len(s.Exemplars) > 0 { + return true + } + } + return false +} + +func hasExemplarsFloatHistograms(samples []record.RefFloatHistogramSampleV2) bool { + for _, s := range samples { + if len(s.Exemplars) > 0 { + return true + } + } + return false +} + func exemplarsForEncoding(es []exemplarWithSeriesRef) []record.RefExemplar { ret := make([]record.RefExemplar, 0, len(es)) for _, e := range es { @@ -1365,6 +1474,114 @@ func handleAppendableError(err error, appended, oooRejected, oobRejected, tooOld // // There are also specific functions to commit histograms and float histograms. func (a *headAppenderBase) commitFloats(b *appendBatch, acc *appenderCommitContext) { + if len(b.floatsV2) > 0 { + var ok, chunkCreated bool + var series *memSeries + + for i, s := range b.floatsV2 { + series = b.floatSeries[i] + series.Lock() + + if value.IsStaleNaN(s.V) { + switch { + case series.lastHistogramValue != nil: + b.histogramsV2 = append(b.histogramsV2, record.RefHistogramSampleV2{ + Ref: series.ref, + ST: s.ST, + T: s.T, + H: &histogram.Histogram{Sum: s.V}, + }) + b.histogramSeries = append(b.histogramSeries, series) + acc.floatsAppended-- + acc.histogramsAppended++ + series.Unlock() + continue + case series.lastFloatHistogramValue != nil: + b.floatHistogramsV2 = append(b.floatHistogramsV2, record.RefFloatHistogramSampleV2{ + Ref: series.ref, + ST: s.ST, + T: s.T, + FH: &histogram.FloatHistogram{Sum: s.V}, + }) + b.floatHistogramSeries = append(b.floatHistogramSeries, series) + acc.floatsAppended-- + acc.histogramsAppended++ + series.Unlock() + continue + } + } + oooSample, _, err := series.appendable(s.T, s.V, a.headMaxt, a.minValidTime, a.oooTimeWindow) + if err != nil { + handleAppendableError(err, &acc.floatsAppended, &acc.floatOOORejected, &acc.floatOOBRejected, &acc.floatTooOldRejected) + } + + prevHeadChunkCount := series.headChunkCount.Load() + switch { + case err != nil: + // Do nothing here. + case oooSample: + var mmapRefs []chunks.ChunkDiskMapperRef + ok, chunkCreated, mmapRefs = series.insert(s.ST, s.T, s.V, nil, nil, acc.appendChunkOpts, acc.oooCapMax, a.head.logger) + if chunkCreated { + r, ok := acc.oooMmapMarkers[series.ref] + if !ok || r != nil { + acc.collectOOORecords(a) + } + + if acc.oooMmapMarkers == nil { + acc.oooMmapMarkers = make(map[chunks.HeadSeriesRef][]chunks.ChunkDiskMapperRef) + } + if len(mmapRefs) > 0 { + acc.oooMmapMarkers[series.ref] = mmapRefs + acc.oooMmapMarkersCount += len(mmapRefs) + } else { + acc.oooMmapMarkers[series.ref] = []chunks.ChunkDiskMapperRef{0} + acc.oooMmapMarkersCount++ + } + } + if ok { + acc.wblSamples = append(acc.wblSamples, record.RefSample{Ref: s.Ref, ST: s.ST, T: s.T, V: s.V}) + if s.T < acc.oooMinT { + acc.oooMinT = s.T + } + if s.T > acc.oooMaxT { + acc.oooMaxT = s.T + } + acc.oooFloatsAccepted++ + } else { + acc.floatsAppended-- + } + default: + wasStale, wasHistogram, oldBuckets := series.sampleState() + isStale := value.IsStaleNaN(s.V) + 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 + } + if s.T > acc.inOrderMaxt { + acc.inOrderMaxt = s.T + } + a.head.updateStaleSeriesMetricOnAppend(wasStale, isStale) + if wasHistogram { + a.head.updateNativeHistogramMetricsOnAppend(true, false, oldBuckets, 0) + } + } else { + acc.floatsAppended-- + } + } + + if chunkCreated { + a.head.onChunkCreated(series, prevHeadChunkCount) + } + + series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow) + a.releasePendingCommit(series) + series.Unlock() + } + return + } + var ok, chunkCreated bool var series *memSeries @@ -1521,6 +1738,86 @@ func (a *headAppenderBase) commitFloats(b *appendBatch, acc *appenderCommitConte // For details on the commitHistograms function, see the commitFloats docs. func (a *headAppenderBase) commitHistograms(b *appendBatch, acc *appenderCommitContext) { + if len(b.histogramsV2) > 0 { + var ok, chunkCreated bool + var series *memSeries + + for i, s := range b.histogramsV2 { + series = b.histogramSeries[i] + series.Lock() + + oooSample, _, err := series.appendableHistogram(s.T, s.H, a.headMaxt, a.minValidTime, a.oooTimeWindow) + if err != nil { + handleAppendableError(err, &acc.histogramsAppended, &acc.histoOOORejected, &acc.histoOOBRejected, &acc.histoTooOldRejected) + } + + prevHeadChunkCount := series.headChunkCount.Load() + switch { + case err != nil: + // Do nothing here. + case oooSample: + var mmapRefs []chunks.ChunkDiskMapperRef + ok, chunkCreated, mmapRefs = series.insert(s.ST, s.T, 0, s.H, nil, acc.appendChunkOpts, acc.oooCapMax, a.head.logger) + if chunkCreated { + r, ok := acc.oooMmapMarkers[series.ref] + if !ok || r != nil { + acc.collectOOORecords(a) + } + + if acc.oooMmapMarkers == nil { + acc.oooMmapMarkers = make(map[chunks.HeadSeriesRef][]chunks.ChunkDiskMapperRef) + } + if len(mmapRefs) > 0 { + acc.oooMmapMarkers[series.ref] = mmapRefs + acc.oooMmapMarkersCount += len(mmapRefs) + } else { + acc.oooMmapMarkers[series.ref] = []chunks.ChunkDiskMapperRef{0} + acc.oooMmapMarkersCount++ + } + } + if ok { + acc.wblHistograms = append(acc.wblHistograms, record.RefHistogramSample{Ref: s.Ref, ST: s.ST, T: s.T, H: s.H}) + if s.T < acc.oooMinT { + acc.oooMinT = s.T + } + if s.T > acc.oooMaxT { + acc.oooMaxT = s.T + } + acc.oooHistogramAccepted++ + } else { + acc.histogramsAppended-- + } + default: + wasStale, wasHistogram, oldBuckets := series.sampleState() + isStale := value.IsStaleNaN(s.H.Sum) + newBuckets := len(s.H.PositiveBuckets) + len(s.H.NegativeBuckets) + ok, chunkCreated = series.appendHistogram(s.ST, s.T, s.H, a.appendID, acc.appendChunkOpts) + if ok { + if s.T < acc.inOrderMint { + acc.inOrderMint = s.T + } + if s.T > acc.inOrderMaxt { + acc.inOrderMaxt = s.T + } + a.head.updateStaleSeriesMetricOnAppend(wasStale, isStale) + a.head.updateNativeHistogramMetricsOnAppend(wasHistogram, true, oldBuckets, newBuckets) + } else { + acc.histogramsAppended-- + acc.histoOOORejected++ + } + } + + if chunkCreated { + a.head.onChunkCreated(series, prevHeadChunkCount) + } + + series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow) + a.releasePendingCommit(series) + series.Unlock() + } + return + } + var ok, chunkCreated bool var series *memSeries @@ -1623,6 +1920,86 @@ func (a *headAppenderBase) commitHistograms(b *appendBatch, acc *appenderCommitC // For details on the commitFloatHistograms function, see the commitFloats docs. func (a *headAppenderBase) commitFloatHistograms(b *appendBatch, acc *appenderCommitContext) { + if len(b.floatHistogramsV2) > 0 { + var ok, chunkCreated bool + var series *memSeries + + for i, s := range b.floatHistogramsV2 { + series = b.floatHistogramSeries[i] + series.Lock() + + oooSample, _, err := series.appendableFloatHistogram(s.T, s.FH, a.headMaxt, a.minValidTime, a.oooTimeWindow) + if err != nil { + handleAppendableError(err, &acc.histogramsAppended, &acc.histoOOORejected, &acc.histoOOBRejected, &acc.histoTooOldRejected) + } + + prevHeadChunkCount := series.headChunkCount.Load() + switch { + case err != nil: + // Do nothing here. + case oooSample: + var mmapRefs []chunks.ChunkDiskMapperRef + ok, chunkCreated, mmapRefs = series.insert(s.ST, s.T, 0, nil, s.FH, acc.appendChunkOpts, acc.oooCapMax, a.head.logger) + if chunkCreated { + r, ok := acc.oooMmapMarkers[series.ref] + if !ok || r != nil { + acc.collectOOORecords(a) + } + + if acc.oooMmapMarkers == nil { + acc.oooMmapMarkers = make(map[chunks.HeadSeriesRef][]chunks.ChunkDiskMapperRef) + } + if len(mmapRefs) > 0 { + acc.oooMmapMarkers[series.ref] = mmapRefs + acc.oooMmapMarkersCount += len(mmapRefs) + } else { + acc.oooMmapMarkers[series.ref] = []chunks.ChunkDiskMapperRef{0} + acc.oooMmapMarkersCount++ + } + } + if ok { + acc.wblFloatHistograms = append(acc.wblFloatHistograms, record.RefFloatHistogramSample{Ref: s.Ref, ST: s.ST, T: s.T, FH: s.FH}) + if s.T < acc.oooMinT { + acc.oooMinT = s.T + } + if s.T > acc.oooMaxT { + acc.oooMaxT = s.T + } + acc.oooHistogramAccepted++ + } else { + acc.histogramsAppended-- + } + default: + wasStale, wasHistogram, oldBuckets := series.sampleState() + isStale := value.IsStaleNaN(s.FH.Sum) + newBuckets := len(s.FH.PositiveBuckets) + len(s.FH.NegativeBuckets) + ok, chunkCreated = series.appendFloatHistogram(s.ST, s.T, s.FH, a.appendID, acc.appendChunkOpts) + if ok { + if s.T < acc.inOrderMint { + acc.inOrderMint = s.T + } + if s.T > acc.inOrderMaxt { + acc.inOrderMaxt = s.T + } + a.head.updateStaleSeriesMetricOnAppend(wasStale, isStale) + a.head.updateNativeHistogramMetricsOnAppend(wasHistogram, true, oldBuckets, newBuckets) + } else { + acc.histogramsAppended-- + acc.histoOOORejected++ + } + } + + if chunkCreated { + a.head.onChunkCreated(series, prevHeadChunkCount) + } + + series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow) + a.releasePendingCommit(series) + series.Unlock() + } + return + } + var ok, chunkCreated bool var series *memSeries @@ -1807,8 +2184,8 @@ func (a *headAppenderBase) Commit() (err error) { } for _, b := range a.batches { - acc.floatsAppended += len(b.floats) - acc.histogramsAppended += len(b.histograms) + len(b.floatHistograms) + acc.floatsAppended += len(b.floats) + len(b.floatsV2) + acc.histogramsAppended += len(b.histograms) + len(b.histogramsV2) + len(b.floatHistograms) + len(b.floatHistogramsV2) a.commitExemplars(b) defer b.close(h) } @@ -2312,24 +2689,20 @@ func (a *headAppenderBase) Rollback() (err error) { h.putTypeMap(a.typesInBatch) }() - var series *memSeries for _, b := range a.batches { - for i := range b.floats { - series = b.floatSeries[i] + for _, series := range b.floatSeries { series.Lock() series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow) a.releasePendingCommit(series) series.Unlock() } - for i := range b.histograms { - series = b.histogramSeries[i] + for _, series := range b.histogramSeries { series.Lock() series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow) a.releasePendingCommit(series) series.Unlock() } - for i := range b.floatHistograms { - series = b.floatHistogramSeries[i] + for _, series := range b.floatHistogramSeries { series.Lock() series.cleanupAppendIDsBelow(a.cleanupAppendIDsBelow) a.releasePendingCommit(series) diff --git a/tsdb/head_append_v2.go b/tsdb/head_append_v2.go index 987ca061f73..a62fdbf9d55 100644 --- a/tsdb/head_append_v2.go +++ b/tsdb/head_append_v2.go @@ -151,14 +151,27 @@ func (a *headAppenderV2) Append(ref storage.SeriesRef, ls labels.Labels, st, t i s = a.bestEffortAppendSTZeroSample(s, ls, st, t, h, fh) } + var attachedExemplars []record.RefExemplar + if len(opts.Exemplars) > 0 { + attachedExemplars = make([]record.RefExemplar, 0, len(opts.Exemplars)) + for _, e := range opts.Exemplars { + attachedExemplars = append(attachedExemplars, record.RefExemplar{ + Ref: s.ref, + T: e.Ts, + V: e.Value, + Labels: e.Labels.WithoutEmpty(), + }) + } + } + var appended *memSeries switch { case fh != nil: isStale = value.IsStaleNaN(fh.Sum) - appended, appErr = a.appendFloatHistogram(s, st, t, fh, opts.RejectOutOfOrder) + appended, appErr = a.appendFloatHistogram(s, st, t, fh, opts.RejectOutOfOrder, attachedExemplars) case h != nil: isStale = value.IsStaleNaN(h.Sum) - appended, appErr = a.appendHistogram(s, st, t, h, opts.RejectOutOfOrder) + appended, appErr = a.appendHistogram(s, st, t, h, opts.RejectOutOfOrder, attachedExemplars) default: isStale = value.IsStaleNaN(v) if isStale { @@ -184,7 +197,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". } - appended, appErr = a.appendFloat(s, st, t, v, opts.RejectOutOfOrder) + appended, appErr = a.appendFloat(s, st, t, v, opts.RejectOutOfOrder, attachedExemplars) } // Handle append error, if any. if appErr != nil { @@ -228,7 +241,7 @@ func (a *headAppenderV2) Append(ref storage.SeriesRef, ls labels.Labels, st, t i // appendFloat appends v to s, and returns the series the sample was appended to, which // may differ from s if s was garbage-collected in the meantime (see lockForAppend). -func (a *headAppenderV2) appendFloat(s *memSeries, st, t int64, v float64, fastRejectOOO bool) (*memSeries, error) { +func (a *headAppenderV2) appendFloat(s *memSeries, st, t int64, v float64, fastRejectOOO bool, exemplars []record.RefExemplar) (*memSeries, error) { s, err := a.lockForAppend(s) if err != nil { return nil, err @@ -252,14 +265,14 @@ func (a *headAppenderV2) appendFloat(s *memSeries, st, t int64, v float64, fastR } b := a.getCurrentBatch(stFloat, s.ref) - b.floats = append(b.floats, record.RefSample{Ref: s.ref, ST: st, T: t, V: v}) + b.floatsV2 = append(b.floatsV2, record.RefSampleV2{Ref: s.ref, ST: st, T: t, V: v, Exemplars: exemplars}) b.floatSeries = append(b.floatSeries, s) return s, nil } // appendHistogram appends h to s, and returns the series the sample was appended to, // which may differ from s if s was garbage-collected in the meantime (see lockForAppend). -func (a *headAppenderV2) appendHistogram(s *memSeries, st, t int64, h *histogram.Histogram, fastRejectOOO bool) (*memSeries, error) { +func (a *headAppenderV2) appendHistogram(s *memSeries, st, t int64, h *histogram.Histogram, fastRejectOOO bool, exemplars []record.RefExemplar) (*memSeries, error) { s, err := a.lockForAppend(s) if err != nil { return nil, err @@ -286,7 +299,7 @@ func (a *headAppenderV2) appendHistogram(s *memSeries, st, t int64, h *histogram sTyp = stCustomBucketHistogram } b := a.getCurrentBatch(sTyp, s.ref) - b.histograms = append(b.histograms, record.RefHistogramSample{Ref: s.ref, ST: st, T: t, H: h}) + b.histogramsV2 = append(b.histogramsV2, record.RefHistogramSampleV2{Ref: s.ref, ST: st, T: t, H: h, Exemplars: exemplars}) b.histogramSeries = append(b.histogramSeries, s) return s, nil } @@ -294,7 +307,7 @@ func (a *headAppenderV2) appendHistogram(s *memSeries, st, t int64, h *histogram // appendFloatHistogram appends fh to s, and returns the series the sample was appended // to, which may differ from s if s was garbage-collected in the meantime (see // lockForAppend). -func (a *headAppenderV2) appendFloatHistogram(s *memSeries, st, t int64, fh *histogram.FloatHistogram, fastRejectOOO bool) (*memSeries, error) { +func (a *headAppenderV2) appendFloatHistogram(s *memSeries, st, t int64, fh *histogram.FloatHistogram, fastRejectOOO bool, exemplars []record.RefExemplar) (*memSeries, error) { s, err := a.lockForAppend(s) if err != nil { return nil, err @@ -321,7 +334,7 @@ func (a *headAppenderV2) appendFloatHistogram(s *memSeries, st, t int64, fh *his sTyp = stCustomBucketFloatHistogram } b := a.getCurrentBatch(sTyp, s.ref) - b.floatHistograms = append(b.floatHistograms, record.RefFloatHistogramSample{Ref: s.ref, ST: st, T: t, FH: fh}) + b.floatHistogramsV2 = append(b.floatHistogramsV2, record.RefFloatHistogramSampleV2{Ref: s.ref, ST: st, T: t, FH: fh, Exemplars: exemplars}) b.floatHistogramSeries = append(b.floatHistogramSeries, s) return s, nil } @@ -356,7 +369,6 @@ func (a *headAppenderV2) appendExemplars(s *memSeries, exemplar []exemplar.Exemp // is implemented. // // ST is an experimental feature, we don't fail the append on errors, just debug log. -// // It returns the series the zero sample was appended to, which may differ from s if s was // garbage-collected in the meantime (see lockForAppend). func (a *headAppenderV2) bestEffortAppendSTZeroSample(s *memSeries, ls labels.Labels, st, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram) *memSeries { @@ -384,7 +396,7 @@ func (a *headAppenderV2) bestEffortAppendSTZeroSample(s *memSeries, ls labels.La ZeroThreshold: fh.ZeroThreshold, CustomValues: fh.CustomValues, } - appended, err = a.appendFloatHistogram(s, 0, st, zeroFloatHistogram, true) + appended, err = a.appendFloatHistogram(s, 0, st, zeroFloatHistogram, true, nil) case h != nil: zeroHistogram := &histogram.Histogram{ // The STZeroSample represents a counter reset by definition. @@ -394,9 +406,9 @@ func (a *headAppenderV2) bestEffortAppendSTZeroSample(s *memSeries, ls labels.La ZeroThreshold: h.ZeroThreshold, CustomValues: h.CustomValues, } - appended, err = a.appendHistogram(s, 0, st, zeroHistogram, true) + appended, err = a.appendHistogram(s, 0, st, zeroHistogram, true, nil) default: - appended, err = a.appendFloat(s, 0, st, 0, true) + appended, err = a.appendFloat(s, 0, st, 0, true, nil) } if err != nil { diff --git a/tsdb/head_append_v2_test.go b/tsdb/head_append_v2_test.go index f043b862964..5d690a8b8f8 100644 --- a/tsdb/head_append_v2_test.go +++ b/tsdb/head_append_v2_test.go @@ -22,7 +22,6 @@ import ( "os" "path" "path/filepath" - "reflect" "slices" "sort" "strconv" @@ -3942,25 +3941,50 @@ func TestWALSampleAndExemplarOrder_AppenderV2(t *testing.T) { lbls := labels.FromStrings("foo", "bar") testcases := map[string]struct { appendF func(app storage.AppenderV2, ts int64) (storage.SeriesRef, error) - expectedType reflect.Type + expectedType record.Type + verifyEx func(t *testing.T, dec record.Decoder, rec []byte) }{ "float sample": { appendF: func(app storage.AppenderV2, ts int64) (storage.SeriesRef, error) { return app.Append(0, lbls, 0, ts, 1.0, nil, nil, storage.AOptions{Exemplars: []exemplar.Exemplar{{Value: 1.0, Ts: 5}}}) }, - expectedType: reflect.TypeFor[[]record.RefSample](), + expectedType: record.SamplesV2, + verifyEx: func(t *testing.T, dec record.Decoder, rec []byte) { + samples, err := dec.SamplesV2(rec, nil) + require.NoError(t, err) + require.Len(t, samples, 1) + require.Len(t, samples[0].Exemplars, 1) + require.Equal(t, 1.0, samples[0].Exemplars[0].V) + require.Equal(t, int64(5), samples[0].Exemplars[0].T) + }, }, "histogram sample": { appendF: func(app storage.AppenderV2, ts int64) (storage.SeriesRef, error) { return app.Append(0, lbls, 0, ts, 0, tsdbutil.GenerateTestHistogram(1), nil, storage.AOptions{Exemplars: []exemplar.Exemplar{{Value: 1.0, Ts: 5}}}) }, - expectedType: reflect.TypeFor[[]record.RefHistogramSample](), + expectedType: record.HistogramSamplesV2, + verifyEx: func(t *testing.T, dec record.Decoder, rec []byte) { + samples, err := dec.HistogramSamplesV2(rec, nil) + require.NoError(t, err) + require.Len(t, samples, 1) + require.Len(t, samples[0].Exemplars, 1) + require.Equal(t, 1.0, samples[0].Exemplars[0].V) + require.Equal(t, int64(5), samples[0].Exemplars[0].T) + }, }, "float histogram sample": { appendF: func(app storage.AppenderV2, ts int64) (storage.SeriesRef, error) { return app.Append(0, lbls, 0, ts, 0, nil, tsdbutil.GenerateTestFloatHistogram(1), storage.AOptions{Exemplars: []exemplar.Exemplar{{Value: 1.0, Ts: 5}}}) }, - expectedType: reflect.TypeFor[[]record.RefFloatHistogramSample](), + expectedType: record.FloatHistogramSamplesV2, + verifyEx: func(t *testing.T, dec record.Decoder, rec []byte) { + samples, err := dec.FloatHistogramSamplesV2(rec, nil) + require.NoError(t, err) + require.Len(t, samples, 1) + require.Len(t, samples[0].Exemplars, 1) + require.Equal(t, 1.0, samples[0].Exemplars[0].V) + require.Equal(t, int64(5), samples[0].Exemplars[0].T) + }, }, } @@ -3977,14 +4001,25 @@ func TestWALSampleAndExemplarOrder_AppenderV2(t *testing.T) { require.NoError(t, app.Commit()) - recs := readTestWAL(t, w.Dir()) - require.Len(t, recs, 3) - _, ok := recs[0].([]record.RefSeries) - require.True(t, ok, "expected first record to be a RefSeries") - actualType := reflect.TypeOf(recs[1]) - require.Equal(t, tc.expectedType, actualType, "expected second record to be a %s", tc.expectedType) - _, ok = recs[2].([]record.RefExemplar) - require.True(t, ok, "expected third record to be a RefExemplar") + sr, err := wlog.NewSegmentsReader(w.Dir()) + require.NoError(t, err) + defer func() { + require.NoError(t, sr.Close()) + }() + + dec := record.NewDecoder(labels.NewSymbolTable(), nil) + r := wlog.NewReader(sr) + + var records [][]byte + for r.Next() { + records = append(records, append([]byte(nil), r.Record()...)) + } + require.NoError(t, r.Err()) + require.Len(t, records, 2, "expected 2 WAL records (RefSeries and compound SamplesV2 with attached exemplar)") + + require.Equal(t, record.Series, dec.Type(records[0]), "expected first record to be Series") + require.Equal(t, tc.expectedType, dec.Type(records[1]), "expected second record to be %v", tc.expectedType) + tc.verifyEx(t, dec, records[1]) }) } } @@ -5218,3 +5253,274 @@ func TestHeadAppenderV2_Histogram_STStorage(t *testing.T) { }) } } + +func TestHeadAppenderV2_CompoundWALRecordsAndExemplarDualWrite(t *testing.T) { + opts := DefaultHeadOptions() + opts.ChunkRange = 100000 + opts.ChunkDirRoot = t.TempDir() + opts.EnableExemplarStorage = true + opts.MaxExemplars.Store(1000) + + h, w := newTestHeadWithOptions(t, compression.None, opts) + defer func() { + require.NoError(t, h.Close()) + }() + + app := h.AppenderV2(context.Background()) + + lsetFloat := labels.FromStrings("__name__", "http_requests_total", "job", "api") + lsetHisto := labels.FromStrings("__name__", "http_request_duration_seconds", "job", "api") + lsetFloatHisto := labels.FromStrings("__name__", "http_request_duration_float_seconds", "job", "api") + + hSample := &histogram.Histogram{ + Schema: 1, + Count: 10, + Sum: 2.5, + ZeroCount: 1, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 0, Length: 1}}, + PositiveBuckets: []int64{9}, + } + + fhSample := &histogram.FloatHistogram{ + Schema: 1, + Count: 10.0, + Sum: 2.5, + ZeroCount: 1.0, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 0, Length: 1}}, + PositiveBuckets: []float64{9.0}, + } + + ex1 := exemplar.Exemplar{Ts: 1000, Value: 1.0, Labels: labels.FromStrings("trace_id", "trace-float-1")} + ex2 := exemplar.Exemplar{Ts: 1000, Value: 2.5, Labels: labels.FromStrings("trace_id", "trace-histo-1")} + ex3 := exemplar.Exemplar{Ts: 1000, Value: 2.5, Labels: labels.FromStrings("trace_id", "trace-floathisto-1")} + + // Append float sample with exemplar + _, err := app.Append(0, lsetFloat, 0, 1000, 1.0, nil, nil, storage.AOptions{ + Exemplars: []exemplar.Exemplar{ex1}, + }) + require.NoError(t, err) + + // Append histogram sample with exemplar + _, err = app.Append(0, lsetHisto, 0, 1000, 0, hSample, nil, storage.AOptions{ + Exemplars: []exemplar.Exemplar{ex2}, + }) + require.NoError(t, err) + + // Append float histogram sample with exemplar + _, err = app.Append(0, lsetFloatHisto, 0, 1000, 0, nil, fhSample, storage.AOptions{ + Exemplars: []exemplar.Exemplar{ex3}, + }) + require.NoError(t, err) + + require.NoError(t, app.Commit()) + + // 1. Verify ExemplarStorage (PromQL query invariant) + exQuerier, err := h.ExemplarQuerier(context.Background()) + require.NoError(t, err) + + res, err := exQuerier.Select(0, 2000, []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests_total")}) + require.NoError(t, err) + require.Len(t, res, 1) + require.Len(t, res[0].Exemplars, 1) + require.Equal(t, ex1.Labels.Map(), res[0].Exemplars[0].Labels.Map()) + require.Equal(t, ex1.Value, res[0].Exemplars[0].Value) + + // 2. Verify WAL Compound Record Encoding + sr, err := wlog.NewSegmentsReader(w.Dir()) + require.NoError(t, err) + defer func() { require.NoError(t, sr.Close()) }() + + dec := record.NewDecoder(labels.NewSymbolTable(), nil) + r := wlog.NewReader(sr) + + var foundSampleV2, foundHistoV2, foundFloatHistoV2 bool + for r.Next() { + rec := r.Record() + switch dec.Type(rec) { + case record.SamplesV2: + samples, err := dec.SamplesV2(rec, nil) + require.NoError(t, err) + for _, s := range samples { + if len(s.Exemplars) > 0 { + foundSampleV2 = true + require.Equal(t, "trace-float-1", s.Exemplars[0].Labels.Get("trace_id")) + } + } + case record.HistogramSamplesV2: + hists, err := dec.HistogramSamplesV2(rec, nil) + require.NoError(t, err) + for _, h := range hists { + if len(h.Exemplars) > 0 { + foundHistoV2 = true + require.Equal(t, "trace-histo-1", h.Exemplars[0].Labels.Get("trace_id")) + } + } + case record.FloatHistogramSamplesV2: + fhists, err := dec.FloatHistogramSamplesV2(rec, nil) + require.NoError(t, err) + for _, fh := range fhists { + if len(fh.Exemplars) > 0 { + foundFloatHistoV2 = true + require.Equal(t, "trace-floathisto-1", fh.Exemplars[0].Labels.Get("trace_id")) + } + } + } + } + require.NoError(t, r.Err()) + require.True(t, foundSampleV2, "expected SamplesV2 record with attached exemplar in WAL") + require.True(t, foundHistoV2, "expected HistogramSamplesV2 record with attached exemplar in WAL") + require.True(t, foundFloatHistoV2, "expected FloatHistogramSamplesV2 record with attached exemplar in WAL") +} + +func TestHeadAppenderV2_RollbackRemovesStagedExemplars(t *testing.T) { + opts := DefaultHeadOptions() + opts.ChunkRange = 100000 + opts.ChunkDirRoot = t.TempDir() + opts.EnableExemplarStorage = true + opts.MaxExemplars.Store(1000) + + h, _ := newTestHeadWithOptions(t, compression.None, opts) + defer func() { + require.NoError(t, h.Close()) + }() + + app := h.AppenderV2(context.Background()) + lset := labels.FromStrings("__name__", "rollback_metric", "job", "test") + ex := exemplar.Exemplar{Ts: 1000, Value: 42.0, Labels: labels.FromStrings("trace_id", "rb-1")} + + _, err := app.Append(0, lset, 0, 1000, 42.0, nil, nil, storage.AOptions{ + Exemplars: []exemplar.Exemplar{ex}, + }) + require.NoError(t, err) + + // Rollback + require.NoError(t, app.Rollback()) + + // Verify ExemplarStorage contains no exemplars + exQuerier, err := h.ExemplarQuerier(context.Background()) + require.NoError(t, err) + res, err := exQuerier.Select(0, 2000, []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "rollback_metric")}) + require.NoError(t, err) + require.Empty(t, res) +} + +func TestHead_MixedVersionWALReplay(t *testing.T) { + dir := t.TempDir() + walDir := filepath.Join(dir, "wal") + w, err := wlog.NewSize(nil, nil, walDir, 32768, compression.None) + require.NoError(t, err) + + var enc record.Encoder + + // 1. Series records + series := []record.RefSeries{ + {Ref: 1, Labels: labels.FromStrings("__name__", "v1_series", "job", "test")}, + {Ref: 2, Labels: labels.FromStrings("__name__", "v2_series", "job", "test")}, + } + require.NoError(t, w.Log(enc.Series(series, nil))) + + // 2. V1 Samples record + v1Samples := []record.RefSample{ + {Ref: 1, T: 1000, V: 10.0}, + {Ref: 1, T: 2000, V: 20.0}, + } + require.NoError(t, w.Log(enc.Samples(v1Samples, nil))) + + // 3. V1 Exemplars record + v1Exemplars := []record.RefExemplar{ + {Ref: 1, T: 1000, V: 10.0, Labels: labels.FromStrings("trace_id", "v1_trace")}, + } + require.NoError(t, w.Log(enc.Exemplars(v1Exemplars, nil))) + + // 4. V2 Samples record (with compound exemplar) + v2Samples := []record.RefSampleV2{ + { + Ref: 2, T: 1000, V: 100.0, + Exemplars: []record.RefExemplar{ + {Ref: 2, T: 1000, V: 100.0, Labels: labels.FromStrings("trace_id", "v2_trace")}, + }, + }, + { + Ref: 2, T: 2000, V: 200.0, + }, + } + require.NoError(t, w.Log(enc.SamplesV2(v2Samples, nil))) + + require.NoError(t, w.Close()) + + // Replay into Head + opts := DefaultHeadOptions() + opts.ChunkRange = 100000 + opts.ChunkDirRoot = dir + opts.EnableExemplarStorage = true + opts.MaxExemplars.Store(1000) + + wal, err := wlog.NewSize(nil, nil, walDir, 32768, compression.None) + require.NoError(t, err) + + h, err := NewHead(nil, nil, wal, nil, opts, nil) + require.NoError(t, err) + defer func() { + require.NoError(t, h.Close()) + }() + + require.NoError(t, h.Init(0)) + + // Verify both series and samples are loaded + q, err := NewBlockQuerier(h, 0, 3000) + require.NoError(t, err) + defer q.Close() + + // Query V1 series + ss1 := q.Select(context.Background(), false, nil, labels.MustNewMatcher(labels.MatchEqual, "__name__", "v1_series")) + require.True(t, ss1.Next()) + s1 := ss1.At() + it1 := s1.Iterator(nil) + require.Equal(t, chunkenc.ValFloat, it1.Next()) + t1, v1 := it1.At() + require.Equal(t, int64(1000), t1) + require.Equal(t, 10.0, v1) + require.Equal(t, chunkenc.ValFloat, it1.Next()) + t2, v2 := it1.At() + require.Equal(t, int64(2000), t2) + require.Equal(t, 20.0, v2) + require.Equal(t, chunkenc.ValNone, it1.Next()) + + // Query V2 series + ss2 := q.Select(context.Background(), false, nil, labels.MustNewMatcher(labels.MatchEqual, "__name__", "v2_series")) + require.True(t, ss2.Next()) + s2 := ss2.At() + it2 := s2.Iterator(nil) + require.Equal(t, chunkenc.ValFloat, it2.Next()) + t3, v3 := it2.At() + require.Equal(t, int64(1000), t3) + require.Equal(t, 100.0, v3) + require.Equal(t, chunkenc.ValFloat, it2.Next()) + t4, v4 := it2.At() + require.Equal(t, int64(2000), t4) + require.Equal(t, 200.0, v4) + require.Equal(t, chunkenc.ValNone, it2.Next()) + + // Verify exemplars from both V1 and V2 records are restored to ExemplarStorage + exQuerier, err := h.ExemplarQuerier(context.Background()) + require.NoError(t, err) + + // V1 exemplar recovery + res1, err := exQuerier.Select(0, 3000, []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "v1_series")}) + require.NoError(t, err) + require.Len(t, res1, 1) + require.Len(t, res1[0].Exemplars, 1) + require.Equal(t, "v1_trace", res1[0].Exemplars[0].Labels.Get("trace_id")) + require.Equal(t, 10.0, res1[0].Exemplars[0].Value) + + // V2 compound exemplar recovery + res2, err := exQuerier.Select(0, 3000, []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "v2_series")}) + require.NoError(t, err) + require.Len(t, res2, 1) + require.Len(t, res2[0].Exemplars, 1) + require.Equal(t, "v2_trace", res2[0].Exemplars[0].Labels.Get("trace_id")) + require.Equal(t, 100.0, res2[0].Exemplars[0].Value) +} + diff --git a/tsdb/head_wal.go b/tsdb/head_wal.go index a4654dccae8..446e2bd1288 100644 --- a/tsdb/head_wal.go +++ b/tsdb/head_wal.go @@ -173,17 +173,47 @@ func (h *Head) loadWAL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch } decoded <- series case record.Samples, record.SamplesV2: - samples := h.wlReplaySamplesPool.Get()[:0] - samples, err = dec.Samples(r.Record(), samples) - if err != nil { - decodeErr = &wlog.CorruptionErr{ - Err: fmt.Errorf("decode samples: %w", err), - Segment: r.Segment(), - Offset: r.Offset(), + rec := r.Record() + if h.opts.EnableExemplarStorage && dec.Type(rec) == record.SamplesV2 { + samplesV2, err := dec.SamplesV2(rec, nil) + if err != nil { + decodeErr = &wlog.CorruptionErr{ + Err: fmt.Errorf("decode samples: %w", err), + Segment: r.Segment(), + Offset: r.Offset(), + } + return } - return + samples := h.wlReplaySamplesPool.Get()[:0] + var exemplars []record.RefExemplar + for _, s := range samplesV2 { + samples = append(samples, record.RefSample{ + Ref: s.Ref, + ST: s.ST, + T: s.T, + V: s.V, + }) + if len(s.Exemplars) > 0 { + exemplars = append(exemplars, s.Exemplars...) + } + } + decoded <- samples + if len(exemplars) > 0 { + decoded <- exemplars + } + } else { + samples := h.wlReplaySamplesPool.Get()[:0] + samples, err = dec.Samples(rec, samples) + if err != nil { + decodeErr = &wlog.CorruptionErr{ + Err: fmt.Errorf("decode samples: %w", err), + Segment: r.Segment(), + Offset: r.Offset(), + } + return + } + decoded <- samples } - decoded <- samples case record.Tombstones: tstones := h.wlReplaytStonesPool.Get()[:0] tstones, err = dec.Tombstones(r.Record(), tstones) @@ -209,29 +239,89 @@ func (h *Head) loadWAL(r *wlog.Reader, syms *labels.SymbolTable, multiRef map[ch } decoded <- exemplars case record.HistogramSamples, record.CustomBucketsHistogramSamples, record.HistogramSamplesV2: - hists := h.wlReplayHistogramsPool.Get()[:0] - hists, err = dec.HistogramSamples(r.Record(), hists) - if err != nil { - decodeErr = &wlog.CorruptionErr{ - Err: fmt.Errorf("decode histograms: %w", err), - Segment: r.Segment(), - Offset: r.Offset(), + rec := r.Record() + if h.opts.EnableExemplarStorage && dec.Type(rec) == record.HistogramSamplesV2 { + histsV2, err := dec.HistogramSamplesV2(rec, nil) + if err != nil { + decodeErr = &wlog.CorruptionErr{ + Err: fmt.Errorf("decode histograms: %w", err), + Segment: r.Segment(), + Offset: r.Offset(), + } + return } - return + hists := h.wlReplayHistogramsPool.Get()[:0] + var exemplars []record.RefExemplar + for _, hs := range histsV2 { + hists = append(hists, record.RefHistogramSample{ + Ref: hs.Ref, + ST: hs.ST, + T: hs.T, + H: hs.H, + }) + if len(hs.Exemplars) > 0 { + exemplars = append(exemplars, hs.Exemplars...) + } + } + decoded <- hists + if len(exemplars) > 0 { + decoded <- exemplars + } + } else { + hists := h.wlReplayHistogramsPool.Get()[:0] + hists, err = dec.HistogramSamples(rec, hists) + if err != nil { + decodeErr = &wlog.CorruptionErr{ + Err: fmt.Errorf("decode histograms: %w", err), + Segment: r.Segment(), + Offset: r.Offset(), + } + return + } + decoded <- hists } - decoded <- hists case record.FloatHistogramSamples, record.CustomBucketsFloatHistogramSamples, record.FloatHistogramSamplesV2: - hists := h.wlReplayFloatHistogramsPool.Get()[:0] - hists, err = dec.FloatHistogramSamples(r.Record(), hists) - if err != nil { - decodeErr = &wlog.CorruptionErr{ - Err: fmt.Errorf("decode float histograms: %w", err), - Segment: r.Segment(), - Offset: r.Offset(), + rec := r.Record() + if h.opts.EnableExemplarStorage && dec.Type(rec) == record.FloatHistogramSamplesV2 { + fhistsV2, err := dec.FloatHistogramSamplesV2(rec, nil) + if err != nil { + decodeErr = &wlog.CorruptionErr{ + Err: fmt.Errorf("decode float histograms: %w", err), + Segment: r.Segment(), + Offset: r.Offset(), + } + return } - return + fhists := h.wlReplayFloatHistogramsPool.Get()[:0] + var exemplars []record.RefExemplar + for _, fhs := range fhistsV2 { + fhists = append(fhists, record.RefFloatHistogramSample{ + Ref: fhs.Ref, + ST: fhs.ST, + T: fhs.T, + FH: fhs.FH, + }) + if len(fhs.Exemplars) > 0 { + exemplars = append(exemplars, fhs.Exemplars...) + } + } + decoded <- fhists + if len(exemplars) > 0 { + decoded <- exemplars + } + } else { + fhists := h.wlReplayFloatHistogramsPool.Get()[:0] + fhists, err = dec.FloatHistogramSamples(rec, fhists) + if err != nil { + decodeErr = &wlog.CorruptionErr{ + Err: fmt.Errorf("decode float histograms: %w", err), + Segment: r.Segment(), + Offset: r.Offset(), + } + return + } + decoded <- fhists } - decoded <- hists case record.Metadata: meta := h.wlReplayMetadataPool.Get()[:0] meta, err := dec.Metadata(r.Record(), meta) diff --git a/tsdb/record/record.go b/tsdb/record/record.go index 046abaa0d45..8692421845a 100644 --- a/tsdb/record/record.go +++ b/tsdb/record/record.go @@ -207,6 +207,38 @@ type RefFloatHistogramSample struct { FH *histogram.FloatHistogram } +// RefSampleV2 is a sample associated with a series reference, start timestamp, timestamp, value, and attached exemplars. +type RefSampleV2 struct { + Ref chunks.HeadSeriesRef + ST, T int64 + V float64 + Exemplars []RefExemplar +} + +// RefHistogramSampleV2 is an integer histogram sample with attached exemplars. +type RefHistogramSampleV2 struct { + Ref chunks.HeadSeriesRef + ST, T int64 + H *histogram.Histogram + Exemplars []RefExemplar +} + +// RefFloatHistogramSampleV2 is a float histogram sample with attached exemplars. +type RefFloatHistogramSampleV2 struct { + Ref chunks.HeadSeriesRef + ST, T int64 + FH *histogram.FloatHistogram + Exemplars []RefExemplar +} + +// RefCustomBucketsHistogramSampleV2 is a custom buckets histogram sample with attached exemplars. +type RefCustomBucketsHistogramSampleV2 struct { + Ref chunks.HeadSeriesRef + ST, T int64 + H *histogram.Histogram + Exemplars []RefExemplar +} + // RefMmapMarker marks that the all the samples of the given series until now have been m-mapped to disk. type RefMmapMarker struct { Ref chunks.HeadSeriesRef @@ -371,6 +403,132 @@ func (*Decoder) samplesV1(dec *encoding.Decbuf, samples []RefSample) ([]RefSampl return samples, nil } +// skipExemplars skips numEx exemplars in dec without allocating heap memory. +func skipExemplars(dec *encoding.Decbuf, numEx int) { + for range numEx { + _ = dec.Varint64() // dtime + _ = dec.Be64() // value + nLabels := dec.Uvarint() + for range nLabels { + l := dec.Uvarint() + dec.Skip(l) + l = dec.Uvarint() + dec.Skip(l) + } + } +} + +// SamplesV2 appends samples with attached exemplars in rec to the given slice. +func (d *Decoder) SamplesV2(rec []byte, samples []RefSampleV2) ([]RefSampleV2, error) { + dec := encoding.Decbuf{B: rec} + switch typ := dec.Byte(); Type(typ) { + case Samples: + return d.samplesV1ToV2(&dec, samples) + case SamplesV2: + return d.samplesV2WithExemplars(&dec, samples) + default: + return nil, fmt.Errorf("invalid record type %v, expected Samples(2) or SamplesV2(11)", typ) + } +} + +func (d *Decoder) samplesV1ToV2(dec *encoding.Decbuf, samples []RefSampleV2) ([]RefSampleV2, error) { + if dec.Len() == 0 { + return samples, nil + } + var ( + baseRef = dec.Be64() + baseTime = dec.Be64int64() + ) + if minSize := dec.Len() / (1 + 1 + 8); cap(samples) < minSize { + samples = make([]RefSampleV2, 0, minSize) + } + for len(dec.B) > 0 && dec.Err() == nil { + dref := dec.Varint64() + dtime := dec.Varint64() + val := dec.Be64() + + samples = append(samples, RefSampleV2{ + Ref: chunks.HeadSeriesRef(int64(baseRef) + dref), + T: baseTime + dtime, + V: math.Float64frombits(val), + }) + } + + if dec.Err() != nil { + return nil, fmt.Errorf("decode error after %d samples: %w", len(samples), dec.Err()) + } + if len(dec.B) > 0 { + return nil, fmt.Errorf("unexpected %d bytes left in entry", len(dec.B)) + } + return samples, nil +} + +func (d *Decoder) samplesV2WithExemplars(dec *encoding.Decbuf, samples []RefSampleV2) ([]RefSampleV2, error) { + if dec.Len() == 0 { + return samples, nil + } + if minSize := len(samples) + dec.Len()/(1+1+8); cap(samples) < minSize { + newSamples := make([]RefSampleV2, len(samples), minSize) + copy(newSamples, samples) + samples = newSamples + } + var firstT, firstST int64 + var prev RefSampleV2 + hasPrev := false + for len(dec.B) > 0 && dec.Err() == nil { + var ref, t, st int64 + var val uint64 + + if !hasPrev { + ref = dec.Varint64() + firstT = dec.Varint64() + t = firstT + st = dec.Varint64() + firstST = st + hasPrev = true + } else { + ref = int64(prev.Ref) + dec.Varint64() + t = firstT + dec.Varint64() + st = readSTMarker(dec, prev.ST, firstST) + } + + val = dec.Be64() + s := RefSampleV2{ + Ref: chunks.HeadSeriesRef(ref), + ST: st, + T: t, + V: math.Float64frombits(val), + } + + numEx := dec.Uvarint() + if numEx > 0 { + s.Exemplars = make([]RefExemplar, 0, numEx) + for range numEx { + dtime := dec.Varint64() + exVal := dec.Be64() + lset := d.DecodeLabels(dec) + s.Exemplars = append(s.Exemplars, RefExemplar{ + Ref: s.Ref, + T: s.T + dtime, + V: math.Float64frombits(exVal), + Labels: lset, + }) + } + } + + prev = s + samples = append(samples, s) + } + + if dec.Err() != nil { + return nil, fmt.Errorf("decode error after %d samples: %w", len(samples), dec.Err()) + } + if len(dec.B) > 0 { + return nil, fmt.Errorf("unexpected %d bytes left in entry", len(dec.B)) + } + return samples, nil +} + // samplesV2 appends samples in rec to the given slice using the V2 algorithm, // which is more efficient and supports ST (See Encoder.samplesV2 definition). func (*Decoder) samplesV2(dec *encoding.Decbuf, samples []RefSample) ([]RefSample, error) { @@ -378,35 +536,46 @@ func (*Decoder) samplesV2(dec *encoding.Decbuf, samples []RefSample) ([]RefSampl return samples, nil } // Allow 1 byte for each varint and 8 for the value; the output slice must be at least that big. - if minSize := dec.Len() / (1 + 1 + 8); cap(samples) < minSize { - samples = make([]RefSample, 0, minSize) + if minSize := len(samples) + dec.Len()/(1+1+8); cap(samples) < minSize { + newSamples := make([]RefSample, len(samples), minSize) + copy(newSamples, samples) + samples = newSamples } var firstT, firstST int64 + var prev RefSample + hasPrev := false for len(dec.B) > 0 && dec.Err() == nil { - var prev RefSample var ref, t, st int64 var val uint64 - if len(samples) == 0 { + if !hasPrev { ref = dec.Varint64() firstT = dec.Varint64() t = firstT st = dec.Varint64() firstST = st + hasPrev = true } else { - prev = samples[len(samples)-1] ref = int64(prev.Ref) + dec.Varint64() t = firstT + dec.Varint64() st = readSTMarker(dec, prev.ST, firstST) } val = dec.Be64() - samples = append(samples, RefSample{ + s := RefSample{ Ref: chunks.HeadSeriesRef(ref), ST: st, T: t, V: math.Float64frombits(val), - }) + } + + numEx := dec.Uvarint() + if numEx > 0 { + skipExemplars(dec, numEx) + } + + prev = s + samples = append(samples, s) } if dec.Err() != nil { @@ -585,6 +754,149 @@ func (d *Decoder) histogramSamplesV1(dec *encoding.Decbuf, histograms []RefHisto return histograms, nil } +// HistogramSamplesV2 appends histogram samples with attached exemplars in rec to the given slice. +func (d *Decoder) HistogramSamplesV2(rec []byte, histograms []RefHistogramSampleV2) ([]RefHistogramSampleV2, error) { + dec := encoding.Decbuf{B: rec} + switch typ := Type(dec.Byte()); typ { + case HistogramSamples, CustomBucketsHistogramSamples: + return d.histogramSamplesV1ToV2(&dec, histograms) + case HistogramSamplesV2: + return d.histogramSamplesV2WithExemplars(&dec, histograms) + default: + return nil, fmt.Errorf("invalid record type %v", typ) + } +} + +func (d *Decoder) histogramSamplesV1ToV2(dec *encoding.Decbuf, histograms []RefHistogramSampleV2) ([]RefHistogramSampleV2, error) { + if dec.Len() == 0 { + return histograms, nil + } + var ( + baseRef = dec.Be64() + baseTime = dec.Be64int64() + ) + for len(dec.B) > 0 && dec.Err() == nil { + dref := dec.Varint64() + dtime := dec.Varint64() + + rh := RefHistogramSampleV2{ + Ref: chunks.HeadSeriesRef(baseRef + uint64(dref)), + T: baseTime + dtime, + H: &histogram.Histogram{}, + } + + DecodeHistogram(dec, rh.H) + + if !histogram.IsKnownSchema(rh.H.Schema) { + d.logger.Warn("skipping histogram with unknown schema in WAL record", "schema", rh.H.Schema, "timestamp", rh.T) + continue + } + if rh.H.Schema > histogram.ExponentialSchemaMax && rh.H.Schema <= histogram.ExponentialSchemaMaxReserved { + if err := rh.H.ReduceResolution(histogram.ExponentialSchemaMax); err != nil { + return nil, fmt.Errorf("error reducing resolution of histogram #%d: %w", len(histograms)+1, err) + } + } + + histograms = append(histograms, rh) + } + + if dec.Err() != nil { + return nil, fmt.Errorf("decode error after %d histograms: %w", len(histograms), dec.Err()) + } + if len(dec.B) > 0 { + return nil, fmt.Errorf("unexpected %d bytes left in entry", len(dec.B)) + } + return histograms, nil +} + +func (d *Decoder) histogramSamplesV2WithExemplars(dec *encoding.Decbuf, histograms []RefHistogramSampleV2) ([]RefHistogramSampleV2, error) { + if dec.Len() == 0 { + return histograms, nil + } + firstRef := chunks.HeadSeriesRef(dec.Varint64()) + firstT := dec.Varint64() + firstST := dec.Varint64() + var ( + prevRef chunks.HeadSeriesRef + prevST int64 + ) + hasPrev := false + + for len(dec.B) > 0 && dec.Err() == nil { + var ref, t, st int64 + if !hasPrev { + ref, t, st = int64(firstRef), firstT, firstST + hasPrev = true + } else { + ref = int64(prevRef) + dec.Varint64() + t = firstT + dec.Varint64() + st = readSTMarker(dec, prevST, firstST) + } + + rh := RefHistogramSampleV2{ + Ref: chunks.HeadSeriesRef(ref), + ST: st, + T: t, + H: &histogram.Histogram{}, + } + prevRef, prevST = rh.Ref, rh.ST + DecodeHistogram(dec, rh.H) + + numEx := dec.Uvarint() + if numEx > 0 { + rh.Exemplars = make([]RefExemplar, 0, numEx) + for range numEx { + dtime := dec.Varint64() + exVal := dec.Be64() + lset := d.DecodeLabels(dec) + rh.Exemplars = append(rh.Exemplars, RefExemplar{ + Ref: rh.Ref, + T: rh.T + dtime, + V: math.Float64frombits(exVal), + Labels: lset, + }) + } + } + + if !histogram.IsKnownSchema(rh.H.Schema) { + d.logger.Warn("skipping histogram with unknown schema in WAL record", "schema", rh.H.Schema, "timestamp", rh.T) + continue + } + if rh.H.Schema > histogram.ExponentialSchemaMax && rh.H.Schema <= histogram.ExponentialSchemaMaxReserved { + if err := rh.H.ReduceResolution(histogram.ExponentialSchemaMax); err != nil { + return nil, fmt.Errorf("error reducing resolution of histogram #%d: %w", len(histograms)+1, err) + } + } + histograms = append(histograms, rh) + } + + if dec.Err() != nil { + return nil, fmt.Errorf("decode error after %d histograms: %w", len(histograms), dec.Err()) + } + if len(dec.B) > 0 { + return nil, fmt.Errorf("unexpected %d bytes left in entry", len(dec.B)) + } + return histograms, nil +} + +// CustomBucketsHistogramSamplesV2 appends custom buckets histogram samples with attached exemplars in rec to the given slice. +func (d *Decoder) CustomBucketsHistogramSamplesV2(rec []byte, histograms []RefCustomBucketsHistogramSampleV2) ([]RefCustomBucketsHistogramSampleV2, error) { + res, err := d.HistogramSamplesV2(rec, nil) + if err != nil { + return nil, err + } + for _, h := range res { + histograms = append(histograms, RefCustomBucketsHistogramSampleV2{ + Ref: h.Ref, + ST: h.ST, + T: h.T, + H: h.H, + Exemplars: h.Exemplars, + }) + } + return histograms, nil +} + // histogramSamplesV2 decodes V2 int-histogram records. func (d *Decoder) histogramSamplesV2(dec *encoding.Decbuf, histograms []RefHistogramSample) ([]RefHistogramSample, error) { if dec.Len() == 0 { @@ -619,6 +931,11 @@ func (d *Decoder) histogramSamplesV2(dec *encoding.Decbuf, histograms []RefHisto prevRef, prevST = rh.Ref, rh.ST DecodeHistogram(dec, rh.H) + numEx := dec.Uvarint() + if numEx > 0 { + skipExemplars(dec, numEx) + } + if !histogram.IsKnownSchema(rh.H.Schema) { d.logger.Warn("skipping histogram with unknown schema in WAL record", "schema", rh.H.Schema, "timestamp", rh.T) continue @@ -759,6 +1076,130 @@ func (d *Decoder) floatHistogramSamplesV1(dec *encoding.Decbuf, histograms []Ref return histograms, nil } +// FloatHistogramSamplesV2 appends float histogram samples with attached exemplars in rec to the given slice. +func (d *Decoder) FloatHistogramSamplesV2(rec []byte, histograms []RefFloatHistogramSampleV2) ([]RefFloatHistogramSampleV2, error) { + dec := encoding.Decbuf{B: rec} + switch typ := Type(dec.Byte()); typ { + case FloatHistogramSamples, CustomBucketsFloatHistogramSamples: + return d.floatHistogramSamplesV1ToV2(&dec, histograms) + case FloatHistogramSamplesV2: + return d.floatHistogramSamplesV2WithExemplars(&dec, histograms) + default: + return nil, fmt.Errorf("invalid record type %v", typ) + } +} + +func (d *Decoder) floatHistogramSamplesV1ToV2(dec *encoding.Decbuf, histograms []RefFloatHistogramSampleV2) ([]RefFloatHistogramSampleV2, error) { + if dec.Len() == 0 { + return histograms, nil + } + var ( + baseRef = dec.Be64() + baseTime = dec.Be64int64() + ) + for len(dec.B) > 0 && dec.Err() == nil { + dref := dec.Varint64() + dtime := dec.Varint64() + + rh := RefFloatHistogramSampleV2{ + Ref: chunks.HeadSeriesRef(baseRef + uint64(dref)), + T: baseTime + dtime, + FH: &histogram.FloatHistogram{}, + } + + DecodeFloatHistogram(dec, rh.FH) + + if !histogram.IsKnownSchema(rh.FH.Schema) { + d.logger.Warn("skipping histogram with unknown schema in WAL record", "schema", rh.FH.Schema, "timestamp", rh.T) + continue + } + if rh.FH.Schema > histogram.ExponentialSchemaMax && rh.FH.Schema <= histogram.ExponentialSchemaMaxReserved { + if err := rh.FH.ReduceResolution(histogram.ExponentialSchemaMax); err != nil { + return nil, fmt.Errorf("error reducing resolution of histogram #%d: %w", len(histograms)+1, err) + } + } + + histograms = append(histograms, rh) + } + + if dec.Err() != nil { + return nil, fmt.Errorf("decode error after %d histograms: %w", len(histograms), dec.Err()) + } + if len(dec.B) > 0 { + return nil, fmt.Errorf("unexpected %d bytes left in entry", len(dec.B)) + } + return histograms, nil +} + +func (d *Decoder) floatHistogramSamplesV2WithExemplars(dec *encoding.Decbuf, histograms []RefFloatHistogramSampleV2) ([]RefFloatHistogramSampleV2, error) { + if dec.Len() == 0 { + return histograms, nil + } + firstRef := chunks.HeadSeriesRef(dec.Varint64()) + firstT := dec.Varint64() + firstST := dec.Varint64() + var prevRef chunks.HeadSeriesRef + var prevST int64 + hasPrev := false + + for len(dec.B) > 0 && dec.Err() == nil { + var ref, t, st int64 + if !hasPrev { + ref, t, st = int64(firstRef), firstT, firstST + hasPrev = true + } else { + ref = int64(prevRef) + dec.Varint64() + t = firstT + dec.Varint64() + st = readSTMarker(dec, prevST, firstST) + } + + rfh := RefFloatHistogramSampleV2{ + Ref: chunks.HeadSeriesRef(ref), + ST: st, + T: t, + FH: &histogram.FloatHistogram{}, + } + prevRef, prevST = rfh.Ref, rfh.ST + DecodeFloatHistogram(dec, rfh.FH) + + numEx := dec.Uvarint() + if numEx > 0 { + rfh.Exemplars = make([]RefExemplar, 0, numEx) + for range numEx { + dtime := dec.Varint64() + exVal := dec.Be64() + lset := d.DecodeLabels(dec) + rfh.Exemplars = append(rfh.Exemplars, RefExemplar{ + Ref: rfh.Ref, + T: rfh.T + dtime, + V: math.Float64frombits(exVal), + Labels: lset, + }) + } + } + + if !histogram.IsKnownSchema(rfh.FH.Schema) { + d.logger.Warn("skipping histogram with unknown schema in WAL record", "schema", rfh.FH.Schema, "timestamp", rfh.T) + continue + } + if rfh.FH.Schema > histogram.ExponentialSchemaMax && rfh.FH.Schema <= histogram.ExponentialSchemaMaxReserved { + if err := rfh.FH.ReduceResolution(histogram.ExponentialSchemaMax); err != nil { + return nil, fmt.Errorf("error reducing resolution of histogram #%d: %w", len(histograms)+1, err) + } + } + + histograms = append(histograms, rfh) + } + + if dec.Err() != nil { + return nil, fmt.Errorf("decode error after %d histograms: %w", len(histograms), dec.Err()) + } + if len(dec.B) > 0 { + return nil, fmt.Errorf("unexpected %d bytes left in entry", len(dec.B)) + } + return histograms, nil +} + // floatHistogramSamplesV2 decodes V2 float-histogram records. func (d *Decoder) floatHistogramSamplesV2(dec *encoding.Decbuf, histograms []RefFloatHistogramSample) ([]RefFloatHistogramSample, error) { if dec.Len() == 0 { @@ -791,6 +1232,11 @@ func (d *Decoder) floatHistogramSamplesV2(dec *encoding.Decbuf, histograms []Ref prevRef, prevST = rfh.Ref, rfh.ST DecodeFloatHistogram(dec, rfh.FH) + numEx := dec.Uvarint() + if numEx > 0 { + skipExemplars(dec, numEx) + } + if !histogram.IsKnownSchema(rfh.FH.Schema) { d.logger.Warn("skipping histogram with unknown schema in WAL record", "schema", rfh.FH.Schema, "timestamp", rfh.T) continue @@ -982,6 +1428,7 @@ func (*Encoder) samplesV2(samples []RefSample, b []byte) []byte { buf.PutVarint64(first.T) buf.PutVarint64(first.ST) buf.PutBE64(math.Float64bits(first.V)) + buf.PutByte(0) // numEx = 0 // Subsequent values are delta to the immediate previous values, and in the // case of start timestamp, use the marker byte to indicate what the value should @@ -995,6 +1442,47 @@ func (*Encoder) samplesV2(samples []RefSample, b []byte) []byte { writeSTMarker(&buf, s.ST, first.ST, prev.ST) buf.PutBE64(math.Float64bits(s.V)) + buf.PutByte(0) // numEx = 0 + } + return buf.Get() +} + +// SamplesV2 appends the encoded samples with attached exemplars to b and returns the resulting slice. +func (*Encoder) SamplesV2(samples []RefSampleV2, b []byte) []byte { + buf := encoding.Encbuf{B: b} + buf.PutByte(byte(SamplesV2)) + + if len(samples) == 0 { + return buf.Get() + } + + first := samples[0] + buf.PutVarint64(int64(first.Ref)) + buf.PutVarint64(first.T) + buf.PutVarint64(first.ST) + buf.PutBE64(math.Float64bits(first.V)) + buf.PutUvarint(len(first.Exemplars)) + for _, ex := range first.Exemplars { + buf.PutVarint64(ex.T - first.T) + buf.PutBE64(math.Float64bits(ex.V)) + EncodeLabels(&buf, ex.Labels) + } + + for i := 1; i < len(samples); i++ { + s := samples[i] + prev := samples[i-1] + + buf.PutVarint64(int64(s.Ref) - int64(prev.Ref)) + buf.PutVarint64(s.T - first.T) + + writeSTMarker(&buf, s.ST, first.ST, prev.ST) + buf.PutBE64(math.Float64bits(s.V)) + buf.PutUvarint(len(s.Exemplars)) + for _, ex := range s.Exemplars { + buf.PutVarint64(ex.T - s.T) + buf.PutBE64(math.Float64bits(ex.V)) + EncodeLabels(&buf, ex.Labels) + } } return buf.Get() } @@ -1134,6 +1622,7 @@ func (*Encoder) histogramSamplesV2(histograms []RefHistogramSample, b []byte) [] buf.PutVarint64(first.ST) prev = first EncodeHistogram(&buf, h.H) + buf.PutByte(0) // numEx = 0 continue } @@ -1142,12 +1631,58 @@ func (*Encoder) histogramSamplesV2(histograms []RefHistogramSample, b []byte) [] writeSTMarker(&buf, h.ST, first.ST, prev.ST) EncodeHistogram(&buf, h.H) + buf.PutByte(0) // numEx = 0 prev = &h } return buf.Get() } +// HistogramSamplesV2 appends the encoded histogram samples with attached exemplars to b and returns the resulting slice. +func (*Encoder) HistogramSamplesV2(histograms []RefHistogramSampleV2, b []byte) []byte { + buf := encoding.Encbuf{B: b} + buf.PutByte(byte(HistogramSamplesV2)) + + if len(histograms) == 0 { + return buf.Get() + } + + var first, prev *RefHistogramSampleV2 + for i := range histograms { + h := &histograms[i] + if first == nil { + first = h + buf.PutVarint64(int64(first.Ref)) + buf.PutVarint64(first.T) + buf.PutVarint64(first.ST) + prev = first + EncodeHistogram(&buf, h.H) + buf.PutUvarint(len(h.Exemplars)) + for _, ex := range h.Exemplars { + buf.PutVarint64(ex.T - h.T) + buf.PutBE64(math.Float64bits(ex.V)) + EncodeLabels(&buf, ex.Labels) + } + continue + } + + buf.PutVarint64(int64(h.Ref) - int64(prev.Ref)) + buf.PutVarint64(h.T - first.T) + + writeSTMarker(&buf, h.ST, first.ST, prev.ST) + EncodeHistogram(&buf, h.H) + buf.PutUvarint(len(h.Exemplars)) + for _, ex := range h.Exemplars { + buf.PutVarint64(ex.T - h.T) + buf.PutBE64(math.Float64bits(ex.V)) + EncodeLabels(&buf, ex.Labels) + } + prev = h + } + + return buf.Get() +} + // CustomBucketsHistogramSamples appends the encoded custom-bucket histogram // samples to b and returns the resulting slice. func (e *Encoder) CustomBucketsHistogramSamples(histograms []RefHistogramSample, b []byte) []byte { @@ -1285,6 +1820,7 @@ func (*Encoder) floatHistogramSamplesV2(histograms []RefFloatHistogramSample, b buf.PutVarint64(first.ST) prev = first EncodeFloatHistogram(&buf, fh.FH) + buf.PutByte(0) // numEx = 0 continue } @@ -1293,12 +1829,73 @@ func (*Encoder) floatHistogramSamplesV2(histograms []RefFloatHistogramSample, b writeSTMarker(&buf, fh.ST, first.ST, prev.ST) EncodeFloatHistogram(&buf, fh.FH) + buf.PutByte(0) // numEx = 0 prev = &fh } return buf.Get() } +// FloatHistogramSamplesV2 appends the encoded float histogram samples with attached exemplars to b and returns the resulting slice. +func (*Encoder) FloatHistogramSamplesV2(histograms []RefFloatHistogramSampleV2, b []byte) []byte { + buf := encoding.Encbuf{B: b} + buf.PutByte(byte(FloatHistogramSamplesV2)) + + if len(histograms) == 0 { + return buf.Get() + } + + var first, prev *RefFloatHistogramSampleV2 + for i := range histograms { + fh := &histograms[i] + if first == nil { + first = fh + buf.PutVarint64(int64(first.Ref)) + buf.PutVarint64(first.T) + buf.PutVarint64(first.ST) + prev = first + EncodeFloatHistogram(&buf, fh.FH) + buf.PutUvarint(len(fh.Exemplars)) + for _, ex := range fh.Exemplars { + buf.PutVarint64(ex.T - fh.T) + buf.PutBE64(math.Float64bits(ex.V)) + EncodeLabels(&buf, ex.Labels) + } + continue + } + + buf.PutVarint64(int64(fh.Ref) - int64(prev.Ref)) + buf.PutVarint64(fh.T - first.T) + + writeSTMarker(&buf, fh.ST, first.ST, prev.ST) + EncodeFloatHistogram(&buf, fh.FH) + buf.PutUvarint(len(fh.Exemplars)) + for _, ex := range fh.Exemplars { + buf.PutVarint64(ex.T - fh.T) + buf.PutBE64(math.Float64bits(ex.V)) + EncodeLabels(&buf, ex.Labels) + } + prev = fh + } + + return buf.Get() +} + +// CustomBucketsHistogramSamplesV2 appends the encoded custom buckets histogram samples with attached exemplars to b and returns the resulting slice. +func (e *Encoder) CustomBucketsHistogramSamplesV2(histograms []RefCustomBucketsHistogramSampleV2, b []byte) []byte { + h2 := make([]RefHistogramSampleV2, len(histograms)) + for i, h := range histograms { + h2[i] = RefHistogramSampleV2{ + Ref: h.Ref, + ST: h.ST, + T: h.T, + H: h.H, + Exemplars: h.Exemplars, + } + } + return e.HistogramSamplesV2(h2, b) +} + // CustomBucketsFloatHistogramSamples appends the encoded custom-bucket float // histogram samples to b and returns the resulting slice. func (e *Encoder) CustomBucketsFloatHistogramSamples(histograms []RefFloatHistogramSample, b []byte) []byte { @@ -1372,3 +1969,48 @@ func EncodeFloatHistogram(buf *encoding.Encbuf, h *histogram.FloatHistogram) { } } } + +// EncodeRefSamplesV2 encodes samples with attached exemplars into byte slice b. +func EncodeRefSamplesV2(samples []RefSampleV2, b []byte) []byte { + var enc Encoder + return enc.SamplesV2(samples, b) +} + +// DecodeRefSamplesV2 decodes samples with attached exemplars from rec into samples. +func DecodeRefSamplesV2(d *Decoder, rec []byte, samples []RefSampleV2) ([]RefSampleV2, error) { + return d.SamplesV2(rec, samples) +} + +// EncodeRefHistogramSamplesV2 encodes histogram samples with attached exemplars into byte slice b. +func EncodeRefHistogramSamplesV2(histograms []RefHistogramSampleV2, b []byte) []byte { + var enc Encoder + return enc.HistogramSamplesV2(histograms, b) +} + +// DecodeRefHistogramSamplesV2 decodes histogram samples with attached exemplars from rec into histograms. +func DecodeRefHistogramSamplesV2(d *Decoder, rec []byte, histograms []RefHistogramSampleV2) ([]RefHistogramSampleV2, error) { + return d.HistogramSamplesV2(rec, histograms) +} + +// EncodeRefFloatHistogramSamplesV2 encodes float histogram samples with attached exemplars into byte slice b. +func EncodeRefFloatHistogramSamplesV2(histograms []RefFloatHistogramSampleV2, b []byte) []byte { + var enc Encoder + return enc.FloatHistogramSamplesV2(histograms, b) +} + +// DecodeRefFloatHistogramSamplesV2 decodes float histogram samples with attached exemplars from rec into histograms. +func DecodeRefFloatHistogramSamplesV2(d *Decoder, rec []byte, histograms []RefFloatHistogramSampleV2) ([]RefFloatHistogramSampleV2, error) { + return d.FloatHistogramSamplesV2(rec, histograms) +} + +// EncodeRefCustomBucketsHistogramSamplesV2 encodes custom buckets histogram samples with attached exemplars into byte slice b. +func EncodeRefCustomBucketsHistogramSamplesV2(histograms []RefCustomBucketsHistogramSampleV2, b []byte) []byte { + var enc Encoder + return enc.CustomBucketsHistogramSamplesV2(histograms, b) +} + +// DecodeRefCustomBucketsHistogramSamplesV2 decodes custom buckets histogram samples with attached exemplars from rec into histograms. +func DecodeRefCustomBucketsHistogramSamplesV2(d *Decoder, rec []byte, histograms []RefCustomBucketsHistogramSampleV2) ([]RefCustomBucketsHistogramSampleV2, error) { + return d.CustomBucketsHistogramSamplesV2(rec, histograms) +} + diff --git a/tsdb/record/record_test.go b/tsdb/record/record_test.go index e2b7bed3bdf..411d7f0ca4f 100644 --- a/tsdb/record/record_test.go +++ b/tsdb/record/record_test.go @@ -1384,3 +1384,357 @@ func BenchmarkDecodeHistogramSamples(b *testing.B) { } } } + +func TestRefSamplesV2(t *testing.T) { + dec := NewDecoder(labels.NewSymbolTable(), promslog.NewNopLogger()) + var enc Encoder + + testCases := []struct { + name string + samples []RefSampleV2 + }{ + { + name: "zero exemplars", + samples: []RefSampleV2{ + {Ref: 1, ST: 100, T: 1000, V: 42.5}, + {Ref: 2, ST: 100, T: 1000, V: 43.5}, + {Ref: 3, ST: 200, T: 1010, V: 44.5}, + }, + }, + { + name: "single exemplar per sample", + samples: []RefSampleV2{ + { + Ref: 10, ST: 0, T: 5000, V: 100.1, + Exemplars: []RefExemplar{ + {Ref: 10, T: 5000, V: 100.1, Labels: labels.FromStrings("trace_id", "abc1234")}, + }, + }, + { + Ref: 11, ST: 0, T: 5000, V: 200.2, + Exemplars: []RefExemplar{ + {Ref: 11, T: 5001, V: 200.2, Labels: labels.FromStrings("trace_id", "def5678", "span_id", "123")}, + }, + }, + }, + }, + { + name: "multiple and mixed exemplars", + samples: []RefSampleV2{ + { + Ref: 100, ST: 1000, T: 2000, V: 1.0, + Exemplars: []RefExemplar{ + {Ref: 100, T: 2000, V: 1.0, Labels: labels.FromStrings("trace_id", "t1")}, + {Ref: 100, T: 2005, V: 1.1, Labels: labels.FromStrings("trace_id", "t2", "env", "prod")}, + }, + }, + { + Ref: 101, ST: 1000, T: 2010, V: 2.0, + Exemplars: nil, + }, + { + Ref: 102, ST: 1000, T: 2020, V: 3.0, + Exemplars: []RefExemplar{ + {Ref: 102, T: 2020, V: 3.0, Labels: labels.FromStrings("trace_id", "t3")}, + }, + }, + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + encoded := enc.SamplesV2(tc.samples, nil) + require.Equal(t, SamplesV2, dec.Type(encoded)) + + // Decode with SamplesV2 (preserving exemplars) + decoded, err := dec.SamplesV2(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(tc.samples), len(decoded)) + for i, exp := range tc.samples { + got := decoded[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.V, got.V) + require.Equal(t, len(exp.Exemplars), len(got.Exemplars)) + for j, exExp := range exp.Exemplars { + exGot := got.Exemplars[j] + require.Equal(t, exExp.Ref, exGot.Ref) + require.Equal(t, exExp.T, exGot.T) + require.Equal(t, exExp.V, exGot.V) + require.Equal(t, exExp.Labels.Map(), exGot.Labels.Map()) + } + } + + // Decode with Samples (zero-alloc stripping of exemplars) + stripped, err := dec.Samples(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(tc.samples), len(stripped)) + for i, exp := range tc.samples { + got := stripped[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.V, got.V) + } + }) + } +} + +func TestRefHistogramsV2(t *testing.T) { + dec := NewDecoder(labels.NewSymbolTable(), promslog.NewNopLogger()) + var enc Encoder + + h1 := &histogram.Histogram{ + Schema: 1, + Count: 50, + Sum: 12.5, + ZeroCount: 5, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 0, Length: 2}}, + PositiveBuckets: []int64{10, 35}, + } + h2 := &histogram.Histogram{ + Schema: 2, + Count: 100, + Sum: 25.0, + ZeroCount: 10, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 1, Length: 2}}, + PositiveBuckets: []int64{20, 70}, + } + + histograms := []RefHistogramSampleV2{ + { + Ref: 50, ST: 100, T: 1000, H: h1, + Exemplars: []RefExemplar{ + {Ref: 50, T: 1000, V: 5.5, Labels: labels.FromStrings("trace_id", "hist_trace_1")}, + {Ref: 50, T: 1002, V: 7.0, Labels: labels.FromStrings("trace_id", "hist_trace_2")}, + }, + }, + { + Ref: 51, ST: 100, T: 1010, H: h2, + Exemplars: nil, + }, + } + + encoded := enc.HistogramSamplesV2(histograms, nil) + require.Equal(t, HistogramSamplesV2, dec.Type(encoded)) + + decoded, err := dec.HistogramSamplesV2(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(histograms), len(decoded)) + for i, exp := range histograms { + got := decoded[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.H.Schema, got.H.Schema) + require.Equal(t, exp.H.Count, got.H.Count) + require.Equal(t, exp.H.Sum, got.H.Sum) + require.Equal(t, len(exp.Exemplars), len(got.Exemplars)) + for j, exExp := range exp.Exemplars { + exGot := got.Exemplars[j] + require.Equal(t, exExp.Ref, exGot.Ref) + require.Equal(t, exExp.T, exGot.T) + require.Equal(t, exExp.V, exGot.V) + require.Equal(t, exExp.Labels.Map(), exGot.Labels.Map()) + } + } + + // Stripped decoding + stripped, err := dec.HistogramSamples(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(histograms), len(stripped)) + for i, exp := range histograms { + got := stripped[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.H.Schema, got.H.Schema) + } +} + +func TestRefFloatHistogramsV2(t *testing.T) { + dec := NewDecoder(labels.NewSymbolTable(), promslog.NewNopLogger()) + var enc Encoder + + fh1 := &histogram.FloatHistogram{ + Schema: 1, + Count: 50.5, + Sum: 12.5, + ZeroCount: 5.5, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 0, Length: 2}}, + PositiveBuckets: []float64{10.2, 34.8}, + } + fh2 := &histogram.FloatHistogram{ + Schema: 2, + Count: 100.0, + Sum: 25.0, + ZeroCount: 10.0, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 1, Length: 2}}, + PositiveBuckets: []float64{20.0, 70.0}, + } + + histograms := []RefFloatHistogramSampleV2{ + { + Ref: 60, ST: 200, T: 2000, FH: fh1, + Exemplars: []RefExemplar{ + {Ref: 60, T: 2000, V: 15.5, Labels: labels.FromStrings("trace_id", "fh_trace_1")}, + }, + }, + { + Ref: 61, ST: 200, T: 2010, FH: fh2, + Exemplars: nil, + }, + } + + encoded := enc.FloatHistogramSamplesV2(histograms, nil) + require.Equal(t, FloatHistogramSamplesV2, dec.Type(encoded)) + + decoded, err := dec.FloatHistogramSamplesV2(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(histograms), len(decoded)) + for i, exp := range histograms { + got := decoded[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.FH.Schema, got.FH.Schema) + require.Equal(t, exp.FH.Count, got.FH.Count) + require.Equal(t, len(exp.Exemplars), len(got.Exemplars)) + for j, exExp := range exp.Exemplars { + exGot := got.Exemplars[j] + require.Equal(t, exExp.Ref, exGot.Ref) + require.Equal(t, exExp.T, exGot.T) + require.Equal(t, exExp.V, exGot.V) + require.Equal(t, exExp.Labels.Map(), exGot.Labels.Map()) + } + } + + // Stripped decoding + stripped, err := dec.FloatHistogramSamples(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(histograms), len(stripped)) + for i, exp := range histograms { + got := stripped[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.FH.Schema, got.FH.Schema) + } +} + +func TestRefCustomBucketsHistogramsV2(t *testing.T) { + dec := NewDecoder(labels.NewSymbolTable(), promslog.NewNopLogger()) + var enc Encoder + + hCustom := &histogram.Histogram{ + Schema: -53, + Count: 10, + Sum: 45.0, + ZeroThreshold: 0.001, + PositiveSpans: []histogram.Span{{Offset: 0, Length: 3}}, + PositiveBuckets: []int64{2, 3, 5}, + CustomValues: []float64{0, 10, 20, 50}, + } + + histograms := []RefCustomBucketsHistogramSampleV2{ + { + Ref: 70, ST: 300, T: 3000, H: hCustom, + Exemplars: []RefExemplar{ + {Ref: 70, T: 3000, V: 9.5, Labels: labels.FromStrings("trace_id", "cb_trace_1")}, + }, + }, + } + + encoded := enc.CustomBucketsHistogramSamplesV2(histograms, nil) + require.Equal(t, HistogramSamplesV2, dec.Type(encoded)) + + decoded, err := dec.CustomBucketsHistogramSamplesV2(encoded, nil) + require.NoError(t, err) + require.Equal(t, len(histograms), len(decoded)) + for i, exp := range histograms { + got := decoded[i] + require.Equal(t, exp.Ref, got.Ref) + require.Equal(t, exp.ST, got.ST) + require.Equal(t, exp.T, got.T) + require.Equal(t, exp.H.Schema, got.H.Schema) + require.Equal(t, exp.H.CustomValues, got.H.CustomValues) + require.Equal(t, len(exp.Exemplars), len(got.Exemplars)) + for j, exExp := range exp.Exemplars { + exGot := got.Exemplars[j] + require.Equal(t, exExp.Ref, exGot.Ref) + require.Equal(t, exExp.T, exGot.T) + require.Equal(t, exExp.V, exGot.V) + require.Equal(t, exExp.Labels.Map(), exGot.Labels.Map()) + } + } +} + +func BenchmarkRecord(b *testing.B) { + const numSamples = 1000 + samples := make([]RefSampleV2, numSamples) + for i := range samples { + samples[i] = RefSampleV2{ + Ref: chunks.HeadSeriesRef(i + 1), + ST: 1000, + T: int64(i)*1000 + 1000, + V: float64(i) * 1.5, + Exemplars: []RefExemplar{ + { + Ref: chunks.HeadSeriesRef(i + 1), + T: int64(i)*1000 + 1000, + V: float64(i) * 1.5, + Labels: labels.FromStrings("trace_id", "abc123456789", "span_id", "987654321"), + }, + }, + } + } + + var enc Encoder + raw := enc.SamplesV2(samples, nil) + dec := NewDecoder(labels.NewSymbolTable(), promslog.NewNopLogger()) + + b.Run("DecodeSamplesV2_ZeroAllocStripping", func(b *testing.B) { + buf := make([]RefSample, 0, numSamples) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + var err error + buf, err = dec.Samples(raw, buf[:0]) + if err != nil { + b.Fatal(err) + } + } + _ = buf + }) + + b.Run("DecodeRefSamplesV2", func(b *testing.B) { + buf := make([]RefSampleV2, 0, numSamples) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + var err error + buf, err = dec.SamplesV2(raw, buf[:0]) + if err != nil { + b.Fatal(err) + } + } + _ = buf + }) + + b.Run("EncodeRefSamplesV2", func(b *testing.B) { + buf := make([]byte, 0, len(raw)) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + buf = enc.SamplesV2(samples, buf[:0]) + } + _ = buf + }) +} + diff --git a/tsdb/wlog/watcher.go b/tsdb/wlog/watcher.go index 196a284d50c..b59040ab93f 100644 --- a/tsdb/wlog/watcher.go +++ b/tsdb/wlog/watcher.go @@ -56,9 +56,12 @@ type WriteTo interface { // // Once returned, the WAL Watcher will not attempt to pass that data again. Append([]record.RefSample) bool + AppendSamplesV2([]record.RefSampleV2) bool AppendExemplars([]record.RefExemplar) bool AppendHistograms([]record.RefHistogramSample) bool + AppendHistogramsV2([]record.RefHistogramSampleV2) bool AppendFloatHistograms([]record.RefFloatHistogramSample) bool + AppendFloatHistogramsV2([]record.RefFloatHistogramSampleV2) bool StoreSeries([]record.RefSeries, int) StoreMetadata([]record.RefMetadata) @@ -515,9 +518,12 @@ func (w *Watcher) garbageCollectSeries(segmentNum int) error { func (w *Watcher) readSegment(r *LiveReader, segmentNum int, tail bool) error { series := w.recordBuf.GetRefSeries(512) samples := w.recordBuf.GetSamples(512) + samplesV2 := make([]record.RefSampleV2, 0, 512) exemplars := w.recordBuf.GetExemplars(512) histograms := w.recordBuf.GetHistograms(512) + histogramsV2 := make([]record.RefHistogramSampleV2, 0, 512) floatHistograms := w.recordBuf.GetFloatHistograms(512) + floatHistogramsV2 := make([]record.RefFloatHistogramSampleV2, 0, 512) metadata := w.recordBuf.GetMetadata(512) defer func() { w.recordBuf.PutRefSeries(series) @@ -549,26 +555,48 @@ func (w *Watcher) readSegment(r *LiveReader, segmentNum int, tail bool) error { if !tail { break } - samples, err = dec.Samples(rec, samples[:0]) - if err != nil { - w.recordDecodeFailsMetric.Inc() - return err - } - // Reuse the underlying array for efficiency. - // It's valid to do, because we override elements that we no longer need to read when filtering. - samplesToSend := samples[:0] - for _, s := range samples { - if s.T > w.startTimestamp { - if !w.sendSamples { - w.sendSamples = true - duration := time.Since(w.startTime) - w.logger.Info("Done replaying WAL", "duration", duration) + if w.sendExemplars { + samplesV2, err = dec.SamplesV2(rec, samplesV2[:0]) + if err != nil { + w.recordDecodeFailsMetric.Inc() + return err + } + samplesToSend := samplesV2[:0] + for _, s := range samplesV2 { + if s.T > w.startTimestamp { + if !w.sendSamples { + w.sendSamples = true + duration := time.Since(w.startTime) + w.logger.Info("Done replaying WAL", "duration", duration) + } + samplesToSend = append(samplesToSend, s) } - samplesToSend = append(samplesToSend, s) } - } - if len(samplesToSend) > 0 { - w.writer.Append(samplesToSend) + if len(samplesToSend) > 0 { + w.writer.AppendSamplesV2(samplesToSend) + } + } else { + samples, err = dec.Samples(rec, samples[:0]) + if err != nil { + w.recordDecodeFailsMetric.Inc() + return err + } + // Reuse the underlying array for efficiency. + // It's valid to do, because we override elements that we no longer need to read when filtering. + samplesToSend := samples[:0] + for _, s := range samples { + if s.T > w.startTimestamp { + if !w.sendSamples { + w.sendSamples = true + duration := time.Since(w.startTime) + w.logger.Info("Done replaying WAL", "duration", duration) + } + samplesToSend = append(samplesToSend, s) + } + } + if len(samplesToSend) > 0 { + w.writer.Append(samplesToSend) + } } case record.Exemplars: @@ -596,26 +624,48 @@ func (w *Watcher) readSegment(r *LiveReader, segmentNum int, tail bool) error { if !tail { break } - histograms, err = dec.HistogramSamples(rec, histograms[:0]) - if err != nil { - w.recordDecodeFailsMetric.Inc() - return err - } - // Reuse the underlying array for efficiency. - // It's valid to do, because we override elements that we no longer need to read when filtering. - histogramsToSend := histograms[:0] - for _, h := range histograms { - if h.T > w.startTimestamp { - if !w.sendSamples { - w.sendSamples = true - duration := time.Since(w.startTime) - w.logger.Info("Done replaying WAL", "duration", duration) + if w.sendExemplars { + histogramsV2, err = dec.HistogramSamplesV2(rec, histogramsV2[:0]) + if err != nil { + w.recordDecodeFailsMetric.Inc() + return err + } + histogramsToSend := histogramsV2[:0] + for _, h := range histogramsV2 { + if h.T > w.startTimestamp { + if !w.sendSamples { + w.sendSamples = true + duration := time.Since(w.startTime) + w.logger.Info("Done replaying WAL", "duration", duration) + } + histogramsToSend = append(histogramsToSend, h) } - histogramsToSend = append(histogramsToSend, h) } - } - if len(histogramsToSend) > 0 { - w.writer.AppendHistograms(histogramsToSend) + if len(histogramsToSend) > 0 { + w.writer.AppendHistogramsV2(histogramsToSend) + } + } else { + histograms, err = dec.HistogramSamples(rec, histograms[:0]) + if err != nil { + w.recordDecodeFailsMetric.Inc() + return err + } + // Reuse the underlying array for efficiency. + // It's valid to do, because we override elements that we no longer need to read when filtering. + histogramsToSend := histograms[:0] + for _, h := range histograms { + if h.T > w.startTimestamp { + if !w.sendSamples { + w.sendSamples = true + duration := time.Since(w.startTime) + w.logger.Info("Done replaying WAL", "duration", duration) + } + histogramsToSend = append(histogramsToSend, h) + } + } + if len(histogramsToSend) > 0 { + w.writer.AppendHistograms(histogramsToSend) + } } case record.FloatHistogramSamples, record.CustomBucketsFloatHistogramSamples, record.FloatHistogramSamplesV2: @@ -626,26 +676,48 @@ func (w *Watcher) readSegment(r *LiveReader, segmentNum int, tail bool) error { if !tail { break } - floatHistograms, err = dec.FloatHistogramSamples(rec, floatHistograms[:0]) - if err != nil { - w.recordDecodeFailsMetric.Inc() - return err - } - // Reuse the underlying array for efficiency. - // It's valid to do, because we override elements that we no longer need to read when filtering. - floatHistogramsToSend := floatHistograms[:0] - for _, fh := range floatHistograms { - if fh.T > w.startTimestamp { - if !w.sendSamples { - w.sendSamples = true - duration := time.Since(w.startTime) - w.logger.Info("Done replaying WAL", "duration", duration) + if w.sendExemplars { + floatHistogramsV2, err = dec.FloatHistogramSamplesV2(rec, floatHistogramsV2[:0]) + if err != nil { + w.recordDecodeFailsMetric.Inc() + return err + } + floatHistogramsToSend := floatHistogramsV2[:0] + for _, fh := range floatHistogramsV2 { + if fh.T > w.startTimestamp { + if !w.sendSamples { + w.sendSamples = true + duration := time.Since(w.startTime) + w.logger.Info("Done replaying WAL", "duration", duration) + } + floatHistogramsToSend = append(floatHistogramsToSend, fh) } - floatHistogramsToSend = append(floatHistogramsToSend, fh) } - } - if len(floatHistogramsToSend) > 0 { - w.writer.AppendFloatHistograms(floatHistogramsToSend) + if len(floatHistogramsToSend) > 0 { + w.writer.AppendFloatHistogramsV2(floatHistogramsToSend) + } + } else { + floatHistograms, err = dec.FloatHistogramSamples(rec, floatHistograms[:0]) + if err != nil { + w.recordDecodeFailsMetric.Inc() + return err + } + // Reuse the underlying array for efficiency. + // It's valid to do, because we override elements that we no longer need to read when filtering. + floatHistogramsToSend := floatHistograms[:0] + for _, fh := range floatHistograms { + if fh.T > w.startTimestamp { + if !w.sendSamples { + w.sendSamples = true + duration := time.Since(w.startTime) + w.logger.Info("Done replaying WAL", "duration", duration) + } + floatHistogramsToSend = append(floatHistogramsToSend, fh) + } + } + if len(floatHistogramsToSend) > 0 { + w.writer.AppendFloatHistograms(floatHistogramsToSend) + } } case record.Metadata: diff --git a/tsdb/wlog/watcher_test.go b/tsdb/wlog/watcher_test.go index 9805b7946be..6e388a5468f 100644 --- a/tsdb/wlog/watcher_test.go +++ b/tsdb/wlog/watcher_test.go @@ -99,6 +99,24 @@ func (wtm *writeToMock) Append(s []record.RefSample) bool { return true } +func (wtm *writeToMock) AppendSamplesV2(s []record.RefSampleV2) bool { + wtm.mu.Lock() + defer wtm.mu.Unlock() + + wtm.sampleAppends++ + for _, sample := range s { + wtm.samplesAppended = append(wtm.samplesAppended, record.RefSample{ + Ref: sample.Ref, + ST: sample.ST, + T: sample.T, + V: sample.V, + }) + wtm.exemplarsAppended = append(wtm.exemplarsAppended, sample.Exemplars...) + } + time.Sleep(wtm.delay) + return true +} + func (wtm *writeToMock) AppendExemplars(e []record.RefExemplar) bool { wtm.mu.Lock() defer wtm.mu.Unlock() @@ -119,6 +137,24 @@ func (wtm *writeToMock) AppendHistograms(h []record.RefHistogramSample) bool { return true } +func (wtm *writeToMock) AppendHistogramsV2(h []record.RefHistogramSampleV2) bool { + wtm.mu.Lock() + defer wtm.mu.Unlock() + + time.Sleep(wtm.delay) + wtm.histogramAppends++ + for _, sample := range h { + wtm.histogramsAppended = append(wtm.histogramsAppended, record.RefHistogramSample{ + Ref: sample.Ref, + ST: sample.ST, + T: sample.T, + H: sample.H, + }) + wtm.exemplarsAppended = append(wtm.exemplarsAppended, sample.Exemplars...) + } + return true +} + func (wtm *writeToMock) AppendFloatHistograms(fh []record.RefFloatHistogramSample) bool { wtm.mu.Lock() defer wtm.mu.Unlock() @@ -129,6 +165,24 @@ func (wtm *writeToMock) AppendFloatHistograms(fh []record.RefFloatHistogramSampl return true } +func (wtm *writeToMock) AppendFloatHistogramsV2(fh []record.RefFloatHistogramSampleV2) bool { + wtm.mu.Lock() + defer wtm.mu.Unlock() + + time.Sleep(wtm.delay) + wtm.floatHistogramsAppends++ + for _, sample := range fh { + wtm.floatHistogramsAppended = append(wtm.floatHistogramsAppended, record.RefFloatHistogramSample{ + Ref: sample.Ref, + ST: sample.ST, + T: sample.T, + FH: sample.FH, + }) + wtm.exemplarsAppended = append(wtm.exemplarsAppended, sample.Exemplars...) + } + return true +} + func (wtm *writeToMock) StoreSeries(series []record.RefSeries, index int) { wtm.mu.Lock() defer wtm.mu.Unlock() @@ -900,3 +954,116 @@ func TestRun_AvoidNotifyWhenBehind(t *testing.T) { } } } + +func TestWALWatcher_CompoundRecordsStreaming(t *testing.T) { + // Test case 1: sendExemplars = true -> captures attached exemplars in AppendSamplesV2 + t.Run("sendExemplars=true", func(t *testing.T) { + overwriteReadTimeout(t, 20*time.Millisecond) + now := time.Now() + ts := timestamp.FromTime(now.Add(1 * time.Second)) + dir := t.TempDir() + wdir := path.Join(dir, "wal") + require.NoError(t, os.Mkdir(wdir, 0o777)) + + w, err := NewSize(nil, nil, wdir, 32768, compression.None) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, w.Close()) + }) + + wt := newWriteToMock(0) + metrics := NewWatcherMetrics(prometheus.NewRegistry()) + watcher := NewWatcher(metrics, nil, nil, "test", wt, dir, true, true, false, nil) + watcher.SetStartTime(now) + watcher.Start() + t.Cleanup(watcher.Stop) + + var enc record.Encoder + // 1. Log Series + series := []record.RefSeries{ + {Ref: 1, Labels: labels.FromStrings("__name__", "http_requests", "job", "api")}, + } + require.NoError(t, w.Log(enc.Series(series, nil))) + + // 2. Log SamplesV2 with attached exemplar + samplesV2 := []record.RefSampleV2{ + { + Ref: 1, ST: 500, T: ts, V: 42.0, + Exemplars: []record.RefExemplar{ + {Ref: 1, T: ts, V: 42.0, Labels: labels.FromStrings("trace_id", "watcher-trace-1")}, + }, + }, + } + require.NoError(t, w.Log(enc.SamplesV2(samplesV2, nil))) + _, err = w.NextSegment() + require.NoError(t, err) + watcher.Notify() + + require.Eventually(t, func() bool { + watcher.Notify() + wt.mu.Lock() + defer wt.mu.Unlock() + return len(wt.samplesAppended) >= 1 && len(wt.exemplarsAppended) >= 1 + }, 10*time.Second, 50*time.Millisecond) + + wt.mu.Lock() + defer wt.mu.Unlock() + require.Equal(t, 1, len(wt.samplesAppended)) + require.Equal(t, 1, len(wt.exemplarsAppended)) + require.Equal(t, "watcher-trace-1", wt.exemplarsAppended[0].Labels.Get("trace_id")) + }) + + // Test case 2: sendExemplars = false -> strips exemplars, calls Append + t.Run("sendExemplars=false", func(t *testing.T) { + overwriteReadTimeout(t, 20*time.Millisecond) + now := time.Now() + ts := timestamp.FromTime(now.Add(1 * time.Second)) + dir := t.TempDir() + wdir := path.Join(dir, "wal") + require.NoError(t, os.Mkdir(wdir, 0o777)) + + w, err := NewSize(nil, nil, wdir, 32768, compression.None) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, w.Close()) + }) + + wt := newWriteToMock(0) + metrics := NewWatcherMetrics(prometheus.NewRegistry()) + watcher := NewWatcher(metrics, nil, nil, "test", wt, dir, false, false, false, nil) + watcher.SetStartTime(now) + watcher.Start() + t.Cleanup(watcher.Stop) + + var enc record.Encoder + series := []record.RefSeries{ + {Ref: 1, Labels: labels.FromStrings("__name__", "http_requests", "job", "api")}, + } + require.NoError(t, w.Log(enc.Series(series, nil))) + + samplesV2 := []record.RefSampleV2{ + { + Ref: 1, ST: 500, T: ts, V: 42.0, + Exemplars: []record.RefExemplar{ + {Ref: 1, T: ts, V: 42.0, Labels: labels.FromStrings("trace_id", "watcher-trace-1")}, + }, + }, + } + require.NoError(t, w.Log(enc.SamplesV2(samplesV2, nil))) + _, err = w.NextSegment() + require.NoError(t, err) + watcher.Notify() + + require.Eventually(t, func() bool { + watcher.Notify() + wt.mu.Lock() + defer wt.mu.Unlock() + return len(wt.samplesAppended) >= 1 + }, 10*time.Second, 50*time.Millisecond) + + wt.mu.Lock() + defer wt.mu.Unlock() + require.Equal(t, 1, len(wt.samplesAppended)) + require.Equal(t, 0, len(wt.exemplarsAppended)) + }) +}