Remote Write

type WriteStorage struct {
    logger *slog.Logger
    reg    prometheus.Registerer
    mtx    sync.Mutex

    watcherMetrics *wlog.WatcherMetrics
    liveReaderMetrics *wlog.LiveReaderMetrics

    // walDir is the TSDB WAL directory being tailed.
    walDir     string
    queues     []*QueueManager
    samplesIn  *ewmaRate  // EMA rate of incoming samples
    flushDeadline time.Duration
    interner   labels.Interner
    ...
}

On ApplyConfig(), WriteStorage diffs the old and new remote_write configs. Unchanged endpoints keep their WAL tail position (no re-replay). New or changed endpoints start a fresh QueueManager from the current WAL head.

The WAL Watcher is a persistent reader that follows the TSDB WAL tail in real-time and decodes records to feed the QueueManager.

// Watcher tails WAL segments, decoding each record type:
func (w *Watcher) readSegment(r *LiveReader, segmentNum int, tail bool) error {
    for r.Next() {
        rec := r.Record()
        switch dec.Type(rec) {
        case record.Series:
            series, err := dec.Series(rec, nil)
            w.writer.StoreSeries(series, segmentNum)  // cache ref→labels

        case record.Samples:
            samples, err := dec.Samples(rec, nil)
            w.writer.Append(samples)  // → QueueManager

        case record.HistogramSamples:
            hSamples, err := dec.HistogramSamples(rec, nil)
            w.writer.AppendHistograms(hSamples)

        case record.MmapMarkers:
            // Head chunks were flushed to disk; update watcher position.
        }
    }
}

The watcher maintains a segmentNum bookmark so it can resume after crashes. When Prometheus restarts, each QueueManager's watcher replays unacknowledged WAL records.

type QueueManager struct {
    logger        *slog.Logger
    cfg           config.QueueConfig
    externalLabels []labels.Label
    relabelConfigs []*relabel.Config

    watcher *wlog.Watcher  // WAL tail reader

    // seriesLabels maps HeadSeriesRef → labels.Labels.
    // Populated when SERIES records are decoded by the watcher.
    seriesLabels   map[chunks.HeadSeriesRef]labels.Labels
    seriesMetadata map[chunks.HeadSeriesRef]*metadata.Metadata
    droppedSeries  map[chunks.HeadSeriesRef]struct{}

    shards    *shards        // current active shards
    numShards int            // dynamically adjusted
    reshardChan chan int      // signals resharding
    ...
}

Dynamic Sharding

The QueueManager monitors queue depth and send rate, then continuously adjusts numShards to saturate the remote endpoint without overloading it:

// Sharding logic (simplified):
//  desired = samplesPerSec * batchSendDeadline / samplesPerBatch
//  clamp to [minShards, maxShards]
//  if desired ≠ numShards: reshard(desired)

Shard → HTTP Batch

1. Each shard maintains a []timeSeries buffer
2. When buffer reaches MaxSamplesPerSend (default 500) or BatchSendDeadline elapses: send
3. Samples serialized to prompb.WriteRequest (protobuf)
4. Compressed with Snappy (RW v1) or Zstd (RW v2)
5. HTTP POST; on 4xx/5xx: exponential backoff and retry
6. On 429 Too Many Requests: respect Retry-After header

Key Queue Config Parameters

// Remote Write v1 request (legacy):
POST /api/v1/write
Content-Type: application/x-protobuf
Content-Encoding: snappy
X-Prometheus-Remote-Write-Version: 0.1.0
Body: snappy( proto.Marshal(prompb.WriteRequest{
    Timeseries: []TimeSeries{
        {Labels: [...], Samples: [{Timestamp, Value}, ...]},
        ...
    },
}))

// Remote Write v2 request (preferred since Prometheus 2.54):
POST /api/v1/write
Content-Type: application/x-protobuf;proto=io.prometheus.write.v2.Request
Content-Encoding: snappy
X-Prometheus-Remote-Write-Version: 2.0.0

The client is configured per remote_write block. It supports TLS, Basic Auth, Bearer tokens, OAuth2, and custom HTTP headers.