upstream/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2026-05-28 04:02:21 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat: RW2 sending ST support

Browse source 
Signed-off-by: bwplotka <bwplotka@gmail.com>

tmp

Signed-off-by: bwplotka <bwplotka@gmail.com>
This commit is contained in:
bwplotka 2026-02-20 10:45:16 +00:00
parent dfd6647002
commit ba1b87f51f
5 changed files with 33 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.github/workflows/ci.yml vendored
View file

@ -147,9 +147,7 @@ jobs:
enable_npm: false
# NOTE: Those tests are based on https://github.com/prometheus/compliance and
# are executed against the ./cmd/prometheus main package.
- run: go test -skip ${SKIP_TESTS} -v --tags=compliance ./compliance/...
env:
SKIP_TESTS: "TestRemoteWriteSender/prometheus/samples/rw2/start_timestamp*" # TODO(bwplotka): PROM-60
- run: go test -v --tags=compliance ./compliance/...
build:
name: Build Prometheus for common architectures

2
compliance/go.mod
View file

@ -2,7 +2,7 @@ module compliance
go 1.25.5
require github.com/prometheus/compliance/remotewrite v0.0.0-20260220101514-bccaa3a70275
require github.com/prometheus/compliance/remotewrite v0.0.0-20260223092825-818283e1171e
require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect

4
compliance/go.sum
View file

@ -30,8 +30,8 @@ github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNw
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.67.2 h1:PcBAckGFTIHt2+L3I33uNRTlKTplNzFctXcWhPyAEN8=
github.com/prometheus/common v0.67.2/go.mod h1:63W3KZb1JOKgcjlIr64WW/LvFGAqKPj0atm+knVGEko=
github.com/prometheus/compliance/remotewrite v0.0.0-20260220101514-bccaa3a70275 h1:NLTtFqM00EuqtisYX9P+hQkjoxNxsR2oUQWDluyD2Xw=
github.com/prometheus/compliance/remotewrite v0.0.0-20260220101514-bccaa3a70275/go.mod h1:VEPZGvpSBbzTKc5acnBj9ng4gfo1DZ4qBsCQnoNFiSc=
github.com/prometheus/compliance/remotewrite v0.0.0-20260223092825-818283e1171e h1:tT/KBv0aSFq4AElo/bSVvUd+yNKj72hkRsyiKU45nIQ=
github.com/prometheus/compliance/remotewrite v0.0.0-20260223092825-818283e1171e/go.mod h1:VEPZGvpSBbzTKc5acnBj9ng4gfo1DZ4qBsCQnoNFiSc=
github.com/prometheus/prometheus v0.307.4-0.20251119130332-1174b0ce4f1f h1:ERPCnBglv9Z4IjkEBTNbcHmZPlryMldXVWLkk7TeBIY=
github.com/prometheus/prometheus v0.307.4-0.20251119130332-1174b0ce4f1f/go.mod h1:7hcXiGf9AXIKW2ehWWzxkvRYJTGmc2StUIJ8mprfxjg=
github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=

44
storage/remote/queue_manager.go
View file

@ -760,11 +760,12 @@ outer:
default:
}
if t.shards.enqueue(s.Ref, timeSeries{
seriesLabels: lbls,
metadata: meta,
timestamp: s.T,
value: s.V,
sType: tSample,
seriesLabels: lbls,
metadata: meta,
startTimestamp: s.ST,
timestamp: s.T,
value: s.V,
sType: tSample,
}) {
continue outer
}
@ -882,9 +883,10 @@ outer:
if t.shards.enqueue(h.Ref, timeSeries{
seriesLabels: lbls,
metadata: meta,
timestamp: h.T,
histogram: h.H,
sType: tHistogram,
// TODO(bwplotka): Populate ST once histogram Ref has it.
timestamp: h.T,
histogram: h.H,
sType: tHistogram,
}) {
continue outer
}
@ -941,8 +943,9 @@ outer:
default:
}
if t.shards.enqueue(h.Ref, timeSeries{
seriesLabels: lbls,
metadata: meta,
seriesLabels: lbls,
metadata: meta,
// TODO(bwplotka): Populate ST once histogram Ref has it.
timestamp: h.T,
floatHistogram: h.FH,
sType: tFloatHistogram,
@ -1396,13 +1399,13 @@ type queue struct {
}
type timeSeries struct {
seriesLabels labels.Labels
value float64
histogram *histogram.Histogram
floatHistogram *histogram.FloatHistogram
metadata *metadata.Metadata
timestamp int64
exemplarLabels labels.Labels
seriesLabels labels.Labels
value float64
histogram *histogram.Histogram
floatHistogram *histogram.FloatHistogram
metadata *metadata.Metadata
startTimestamp, timestamp int64
exemplarLabels labels.Labels
// The type of series: sample, exemplar, or histogram.
sType seriesType
}
@ -1995,8 +1998,9 @@ func populateV2TimeSeries(symbolTable *writev2.SymbolsTable, batch []timeSeries,
switch d.sType {
case tSample:
pendingData[nPending].Samples = append(pendingData[nPending].Samples, writev2.Sample{
Value: d.value,
Timestamp: d.timestamp,
Value: d.value,
Timestamp: d.timestamp,
StartTimestamp: d.startTimestamp,
})
nPendingSamples++
case tExemplar:
@ -2007,9 +2011,11 @@ func populateV2TimeSeries(symbolTable *writev2.SymbolsTable, batch []timeSeries,
})
nPendingExemplars++
case tHistogram:
// TODO(bwplotka): Extend with ST once histograms populate it.
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, writev2.FromIntHistogram(d.timestamp, d.histogram))
nPendingHistograms++
case tFloatHistogram:
// TODO(bwplotka): Extend with ST once histograms populate it.
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, writev2.FromFloatHistogram(d.timestamp, d.floatHistogram))
nPendingHistograms++
case tMetadata:

5
storage/remote/queue_manager_test.go
View file

@ -866,9 +866,12 @@ func generateRecords(c recCase) (ret records) {
Help: "help text",
}
for j := range c.samplesPerSeries {
ts := c.tsFn(i, j)
st := ts - 1 // Keep ST simple for now; we don't need to
// test exact semantics.
ret.samples[i*c.samplesPerSeries+j] = record.RefSample{
Ref: chunks.HeadSeriesRef(i),
T: c.tsFn(i, j),
T: st,
V: float64(i),
}
}