Struct QueueManager

storage/remote/queue_manager.go:421–464

QueueManager manages a queue of samples to be sent to the Storage indicated by the provided WriteClient. Implements writeTo interface used by WAL Watcher.

// indicated by the provided WriteClient. Implements writeTo interface
// used by WAL Watcher.
type QueueManager struct {
	lastSendTimestamp            atomic.Int64
	buildRequestLimitTimestamp   atomic.Int64
	reshardDisableStartTimestamp atomic.Int64 // Time that reshard was disabled.
	reshardDisableEndTimestamp   atomic.Int64 // Time that reshard is disabled until.

	logger                  *slog.Logger
	flushDeadline           time.Duration
	cfg                     config.QueueConfig
	mcfg                    config.MetadataConfig
	externalLabels          []labels.Label
	relabelConfigs          []*relabel.Config
	sendExemplars           bool
	sendNativeHistograms    bool
	enableTypeAndUnitLabels bool
	watcher                 *wlog.Watcher
	metadataWatcher         *MetadataWatcher

	clientMtx   sync.RWMutex
	storeClient WriteClient
	protoMsg    remoteapi.WriteMessageType
	compr       compression.Type

	seriesMtx      sync.Mutex // Covers seriesLabels, seriesMetadata, droppedSeries and builder.
	seriesLabels   map[chunks.HeadSeriesRef]labels.Labels
	seriesMetadata map[chunks.HeadSeriesRef]*metadata.Metadata
	droppedSeries  map[chunks.HeadSeriesRef]struct{}
	builder        *labels.Builder

	seriesSegmentMtx     sync.Mutex // Covers seriesSegmentIndexes - if you also lock seriesMtx, take seriesMtx first.
	seriesSegmentIndexes map[chunks.HeadSeriesRef]int

	shards      *shards
	numShards   int
	reshardChan chan int
	quit        chan struct{}
	wg          sync.WaitGroup

	dataIn, dataDropped, dataOut, dataOutDuration *ewmaRate

	metrics              *queueManagerMetrics
	interner             *pool
	highestRecvTimestamp *maxTimestamp
}

// NewQueueManager builds a new QueueManager and starts a new
// WAL watcher with queue manager as the WriteTo destination.