Function NewQueueManager

storage/remote/queue_manager.go:471–558

NewQueueManager builds a new QueueManager and starts a new WAL watcher with queue manager as the WriteTo destination. The WAL watcher takes the dir parameter as the base directory for where the WAL shall be located. Note that the full path to the WAL directory will be constructed as /wal.

(
	metrics *queueManagerMetrics,
	watcherMetrics *wlog.WatcherMetrics,
	readerMetrics *wlog.LiveReaderMetrics,
	logger *slog.Logger,
	dir string,
	samplesIn *ewmaRate,
	cfg config.QueueConfig,
	mCfg config.MetadataConfig,
	externalLabels labels.Labels,
	relabelConfigs []*relabel.Config,
	client WriteClient,
	flushDeadline time.Duration,
	interner *pool,
	highestRecvTimestamp *maxTimestamp,
	sm ReadyScrapeManager,
	enableExemplarRemoteWrite bool,
	enableNativeHistogramRemoteWrite bool,
	enableTypeAndUnitLabels bool,
	protoMsg remoteapi.WriteMessageType,
	recordBuf *record.BuffersPool,
)

// for where the WAL shall be located. Note that the full path to
// the WAL directory will be constructed as <dir>/wal.
func NewQueueManager(
	metrics *queueManagerMetrics,
	watcherMetrics *wlog.WatcherMetrics,
	readerMetrics *wlog.LiveReaderMetrics,
	logger *slog.Logger,
	dir string,
	samplesIn *ewmaRate,
	cfg config.QueueConfig,
	mCfg config.MetadataConfig,
	externalLabels labels.Labels,
	relabelConfigs []*relabel.Config,
	client WriteClient,
	flushDeadline time.Duration,
	interner *pool,
	highestRecvTimestamp *maxTimestamp,
	sm ReadyScrapeManager,
	enableExemplarRemoteWrite bool,
	enableNativeHistogramRemoteWrite bool,
	enableTypeAndUnitLabels bool,
	protoMsg remoteapi.WriteMessageType,
	recordBuf *record.BuffersPool,
) *QueueManager {
	if logger == nil {
		logger = promslog.NewNopLogger()
	}

	// Copy externalLabels into a slice, which we need for processExternalLabels.
	extLabelsSlice := make([]labels.Label, 0, externalLabels.Len())
	externalLabels.Range(func(l labels.Label) {
		extLabelsSlice = append(extLabelsSlice, l)
	})

	logger = logger.With(remoteName, client.Name(), endpoint, client.Endpoint())
	t := &QueueManager{
		logger:                  logger,
		flushDeadline:           flushDeadline,
		cfg:                     cfg,
		mcfg:                    mCfg,
		externalLabels:          extLabelsSlice,
		relabelConfigs:          relabelConfigs,
		storeClient:             client,
		sendExemplars:           enableExemplarRemoteWrite,
		sendNativeHistograms:    enableNativeHistogramRemoteWrite,
		enableTypeAndUnitLabels: enableTypeAndUnitLabels,

		seriesLabels:         make(map[chunks.HeadSeriesRef]labels.Labels),
		seriesMetadata:       make(map[chunks.HeadSeriesRef]*metadata.Metadata),
		seriesSegmentIndexes: make(map[chunks.HeadSeriesRef]int),
		droppedSeries:        make(map[chunks.HeadSeriesRef]struct{}),
		builder:              labels.NewBuilder(labels.EmptyLabels()),

		numShards:   cfg.MinShards,
		reshardChan: make(chan int),
		quit:        make(chan struct{}),

		dataIn:          samplesIn,
		dataDropped:     newEWMARate(ewmaWeight, shardUpdateDuration),
		dataOut:         newEWMARate(ewmaWeight, shardUpdateDuration),