Method updateQueueLists

modules/distributor/forwarder/manager.go:144–176

()

}

func (m *Manager) updateQueueLists() {
	m.tenantToQueueListMu.Lock()
	defer m.tenantToQueueListMu.Unlock()

	queueListsToAdd := make(map[string]*queueList)
	for tenantID, ql := range m.tenantToQueueList {
		forwarderNames := m.overrides.Forwarders(tenantID)
		if len(forwarderNames) < 1 {
			go m.shutdownQueueList(tenantID, ql)
			delete(m.tenantToQueueList, tenantID)

			continue
		}

		if ql.shouldUpdate(forwarderNames) {
			go m.shutdownQueueList(tenantID, ql)
			delete(m.tenantToQueueList, tenantID)

			newQl, err := newQueueList(m.logger, tenantID, forwarderNames, m.forwarderNameToForwarder)
			if err != nil {
				_ = level.Warn(m.logger).Log("msg", "failed to create queue list", "err", err)

				continue
			}

			queueListsToAdd[tenantID] = newQl
		}
	}

	for tenantID, ql := range queueListsToAdd {
		m.tenantToQueueList[tenantID] = ql
	}
}

func (m *Manager) shutdown() error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)