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

Closed
dashpole wants to merge 2 commits into from
Closed

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) {

@dashpole dashpole Sep 8, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

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

The WAL emits a whole samples record and then a whole exemplars record (head_append.go:1141 then :1187, dispatched at watcher.go:571 then :589), so by the time an exemplar reaches Append its sample has usually already been flushed out of q.batch — and the ring only helps the exemplar-first order, which never happens. Driving queue.Append with default max_samples_per_send: 2000 and one shard: 500 series → 100% of exemplars delivered, 5,000 → 20%, 20,000 → 0%; the ring delivered zero in all three. The correlation buffer needs to hold samples across the batch flush, not exemplars.

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.

@dashpole dashpole Sep 8, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

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

TryAttachMatchingExemplars at line 1504 has already tombstoned and removed the matched exemplars from the ring by the time this branch un-appends the datum and returns false. datum is a local copy, so those exemplars are gone — not sent, and not counted by onDrop, so unmatched_exemplars_dropped_total misses them while pendingExemplars stays permanently elevated. I reproduced it with a 2/2 queue: pending goes 1 → 0, onDrop calls = 0, and the retried sample carries 0 exemplars. It is latent today because the ring path is itself unreachable (F1), but it would become the main leak as soon as F1 is fixed — please only commit the ring removal once the append has succeeded.

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