Skip to content

QueueManager Shard-Level Coalescing #5

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Draft
dashpole wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

QueueManager Shard-Level Coalescing #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e475f5e
remote: implement QueueManager shard-level coalescing window for PRW 2.0
dashpole Aug 27, 2026
964f989
Address review feedback: shard coalescer cycle bounds, index preserva…
dashpole Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 136 additions & 12 deletions storage/remote/queue_manager.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
"go.uber.org/atomic"

"github.com/prometheus/prometheus/config"
"github.com/prometheus/prometheus/model/exemplar"
"github.com/prometheus/prometheus/model/histogram"
"github.com/prometheus/prometheus/model/labels"
"github.com/prometheus/prometheus/model/metadata"
Expand Down Expand Up @@ -69,7 +70,7 @@
type queueManagerMetrics struct {
reg prometheus.Registerer

samplesTotal prometheus.Counter

Check failure on line 73 in storage/remote/queue_manager.go

View workflow job for this annotation

GitHub Actions / golangci-lint 

File is not properly formatted (gci)
exemplarsTotal prometheus.Counter
histogramsTotal prometheus.Counter
metadataTotal prometheus.Counter
Expand All @@ -96,9 +97,10 @@
maxNumShards prometheus.Gauge
minNumShards prometheus.Gauge
desiredNumShards prometheus.Gauge
sentBytesTotal prometheus.Counter
metadataBytesTotal prometheus.Counter
maxSamplesPerSend prometheus.Gauge
sentBytesTotal prometheus.Counter
metadataBytesTotal prometheus.Counter
maxSamplesPerSend prometheus.Gauge
unmatchedExemplarsDroppedTotal prometheus.Counter
}

func newQueueManagerMetrics(r prometheus.Registerer, rn, e string) *queueManagerMetrics {
Expand Down Expand Up @@ -328,6 +330,13 @@
Help: "The maximum number of samples to be sent, in a single request, to the remote storage. Note that, when sending of exemplars over remote write is enabled, exemplars count towards this limit.",
ConstLabels: constLabels,
})
m.unmatchedExemplarsDroppedTotal = prometheus.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystem,
Name: "unmatched_exemplars_dropped_total",
Help: "Total number of exemplars dropped due to not matching any sample within the coalescing window.",
ConstLabels: constLabels,
})

return m
}
Expand Down Expand Up @@ -365,6 +374,7 @@
m.sentBytesTotal,
m.metadataBytesTotal,
m.maxSamplesPerSend,
m.unmatchedExemplarsDroppedTotal,
)
}
}
Expand Down Expand Up @@ -401,6 +411,7 @@
m.reg.Unregister(m.sentBytesTotal)
m.reg.Unregister(m.metadataBytesTotal)
m.reg.Unregister(m.maxSamplesPerSend)
m.reg.Unregister(m.unmatchedExemplarsDroppedTotal)
}
}

Expand Down Expand Up @@ -1292,8 +1303,25 @@
s.qm.metrics.numShards.Set(float64(n))

newQueues := make([]*queue, n)
onDrop := func(ex exemplar.Exemplar) {
s.enqueuedExemplars.Sub(1)
if s.qm.metrics != nil {
if s.qm.metrics.unmatchedExemplarsDroppedTotal != nil {
s.qm.metrics.unmatchedExemplarsDroppedTotal.Inc()
}
if s.qm.metrics.droppedExemplarsTotal != nil {
s.qm.metrics.droppedExemplarsTotal.WithLabelValues("unmatched").Inc()
}
if s.qm.metrics.pendingExemplars != nil {
s.qm.metrics.pendingExemplars.Dec()
}
}
if s.qm.dataDropped != nil {
s.qm.dataDropped.incr(1)
}
}
for i := range n {
newQueues[i] = newQueue(s.qm.cfg.MaxSamplesPerSend, s.qm.cfg.Capacity)
newQueues[i] = newQueue(s.qm.cfg.MaxSamplesPerSend, s.qm.cfg.Capacity, s.qm.protoMsg, onDrop)
}

s.queues = newQueues
Expand Down Expand Up @@ -1367,6 +1395,7 @@
case <-s.softShutdown:
return false
default:
data.seriesRef = ref
appended := s.queues[shard].Append(data)
if !appended {
return false
Expand Down Expand Up @@ -1395,6 +1424,9 @@
batch []timeSeries
batchQueue chan []timeSeries

coalescer *shardCoalescer
protoMsg remoteapi.WriteMessageType

// Since we know there are a limited number of batches out, using a stack
// is easy and safe so a sync.Pool is not necessary.
// poolMtx covers adding and removing batches from the batchPool.
Expand All @@ -1403,13 +1435,15 @@
}

type timeSeries struct {
seriesRef chunks.HeadSeriesRef
seriesLabels labels.Labels
value float64
histogram *histogram.Histogram
floatHistogram *histogram.FloatHistogram
metadata *metadata.Metadata
startTimestamp, timestamp int64
exemplarLabels labels.Labels
exemplars []exemplar.Exemplar
// The type of series: sample, exemplar, or histogram.
sType seriesType
}
Expand All @@ -1424,7 +1458,7 @@
tMetadata
)

func newQueue(batchSize, capacity int) *queue {
func newQueue(batchSize, capacity int, protoMsg remoteapi.WriteMessageType, onDrop func(exemplar.Exemplar)) *queue {
batches := capacity / batchSize
// Always create an unbuffered channel even if capacity is configured to be
// less than max_samples_per_send.
Expand All @@ -1437,6 +1471,8 @@
// batchPool should have capacity for everything in the channel + 1 for
// the batch being processed.
batchPool: make([][]timeSeries, 0, batches+1),
coalescer: newShardCoalescer(defaultRingBufferSize, onDrop),
protoMsg: protoMsg,
}
}

Expand All @@ -1445,6 +1481,32 @@
func (q *queue) Append(datum timeSeries) bool {
q.batchMtx.Lock()
defer q.batchMtx.Unlock()

if q.protoMsg == remoteapi.WriteV2MessageType && q.coalescer != nil {
if datum.sType == tExemplar {
ex := exemplar.Exemplar{
Labels: datum.exemplarLabels,
Value: datum.value,
Ts: datum.timestamp,
HasTs: true,
}
// Try attaching to an existing un-flushed sample/histogram in the current batch.
if q.coalescer.TryAttachToBatch(q.batch, datum.seriesRef, ex) {
return true
}
// Buffer in ring buffer.
q.coalescer.AddPendingExemplar(datum.seriesRef, ex)
return true
}

if datum.sType == tSample || datum.sType == tHistogram || datum.sType == tFloatHistogram {
// Check if there are any matching pending exemplars in the ring buffer.
if matched := q.coalescer.TryAttachMatchingExemplars(datum.seriesRef, datum.timestamp); len(matched) > 0 {
datum.exemplars = append(datum.exemplars, matched...)
}
}
}

// TODO(cstyan): Check if metadata now means we've reduced the total # of samples
// we can batch together here, and if so find a way to not include metadata
// in the batch size calculation.
Expand Down Expand Up @@ -1505,6 +1567,9 @@

q.batchMtx.Lock()
defer q.batchMtx.Unlock()
if q.coalescer != nil {
q.coalescer.FlushAndClear()
}
q.batch = nil
close(q.batchQueue)
}
Expand Down Expand Up @@ -1604,7 +1669,7 @@
_ = s.sendSamples(ctx, pendingData[:n], nPendingSamples, nPendingExemplars, nPendingHistograms, pBuf, encBuf, compr)
case remoteapi.WriteV2MessageType:
nPendingSamples, nPendingExemplars, nPendingHistograms, nPendingMetadata, nUnexpectedMetadata := populateV2TimeSeries(&symbolTable, batch, pendingDataV2, s.qm.sendExemplars, s.qm.sendNativeHistograms, s.qm.enableTypeAndUnitLabels)
n := nPendingSamples + nPendingExemplars + nPendingHistograms
n := len(batch)
if nUnexpectedMetadata > 0 {
s.qm.logger.Warn("unexpected metadata sType in populateV2TimeSeries", "count", nUnexpectedMetadata)
}
Expand Down Expand Up @@ -1678,6 +1743,15 @@
Value: d.value,
Timestamp: d.timestamp,
})
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.Value,
Timestamp: ex.Ts,
})
}
}
nPendingSamples++
case tExemplar:
pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, prompb.Exemplar{
Expand All @@ -1688,9 +1762,27 @@
nPendingExemplars++
case tHistogram:
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, prompb.FromIntHistogram(d.timestamp, d.histogram))
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.Value,
Timestamp: ex.Ts,
})
}
}
nPendingHistograms++
case tFloatHistogram:
pendingData[nPending].Histograms = append(pendingData[nPending].Histograms, prompb.FromFloatHistogram(d.timestamp, d.floatHistogram))
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.Value,
Timestamp: ex.Ts,
})
}
}
nPendingHistograms++
}
}
Expand Down Expand Up @@ -2017,19 +2109,51 @@
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.Value,
Timestamp: ex.Ts,
})
nPendingExemplars++
}
}
case tExemplar:
pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, writev2.Exemplar{
LabelsRefs: symbolTable.SymbolizeLabels(d.exemplarLabels, nil), // TODO: optimize, reuse slice
Value: d.value,
Timestamp: d.timestamp,
})
nPendingExemplars++
if sendExemplars {
pendingData[nPending].Exemplars = append(pendingData[nPending].Exemplars, writev2.Exemplar{
LabelsRefs: symbolTable.SymbolizeLabels(d.exemplarLabels, nil), // TODO: optimize, reuse slice
Value: d.value,
Timestamp: d.timestamp,
})
nPendingExemplars++
}
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.Value,
Timestamp: ex.Ts,
})
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.Value,
Timestamp: ex.Ts,
})
nPendingExemplars++
}
}
case tMetadata:
nUnexpectedMetadata++
}
Expand Down
Loading
Loading