Function newQueueList

modules/distributor/forwarder/manager.go:218–251

(logger log.Logger, tenantID string, forwarderNames []string, forwarderNameToForwarder map[string]Forwarder)

}

func newQueueList(logger log.Logger, tenantID string, forwarderNames []string, forwarderNameToForwarder map[string]Forwarder) (*queueList, error) {
	forwarderNameToQueue := make(map[string]*queue.Queue[ptrace.Traces], len(forwarderNames))
	list := make(List, 0, len(forwarderNames))
	for _, forwarderName := range forwarderNames {
		forwarder, found := forwarderNameToForwarder[forwarderName]
		if !found {
			return nil, fmt.Errorf("failed to find forwarder by name: forwarderName=%s, tenantID=%s", forwarderName, tenantID)
		}

		queueCfg := queue.Config{
			Name:        forwarderName,
			TenantID:    tenantID,
			Size:        defaultQueueSize,
			WorkerCount: defaultWorkerCount,
		}

		processFunc := func(ctx context.Context, traces ptrace.Traces) {
			if err := forwarder.ForwardTraces(ctx, traces); err != nil {
				_ = level.Warn(logger).Log("msg", "failed to forward batches", "forwarderName", forwarderName, "tenantID", tenantID, "err", err)
			}
		}
		newQueue := queue.New(queueCfg, logger, processFunc)
		newQueue.StartWorkers()
		forwarderNameToQueue[forwarderName] = newQueue
		list = append(list, queueAdapter{queue: newQueue})
	}

	return &queueList{
		logger:               logger,
		tenantID:             tenantID,
		forwarderNameToQueue: forwarderNameToQueue,
		list:                 list,
	}, nil
}

func (l *queueList) shouldUpdate(forwarderNames []string) bool {
	if len(forwarderNames) != len(l.forwarderNameToQueue) {