Method Push

modules/distributor/queue/queue.go:80–105

Push pushes data onto a queue. If the queue is full, the data is dropped

(ctx context.Context, data T)

// Push pushes data onto a queue.
// If the queue is full, the data is dropped
func (m *Queue[T]) Push(ctx context.Context, data T) error {
	if m.readOnly.Load() {
		return fmt.Errorf("queue is read-only")
	}

	m.pushesTotalMetrics.WithLabelValues(m.name, m.tenantID).Inc()

	select {
	case <-ctx.Done():
		m.pushesFailuresTotalMetrics.WithLabelValues(m.name, m.tenantID).Inc()
		return fmt.Errorf("failed to push data to queue for tenant=%s and queue_name=%s: %w", m.tenantID, m.name, ctx.Err())
	default:
	}

	select {
	case m.reqChan <- data:
		m.lengthMetric.WithLabelValues(m.name, m.tenantID).Inc()
		return nil
	default:
	}

	// Fail fast if the queue is full
	m.pushesFailuresTotalMetrics.WithLabelValues(m.name, m.tenantID).Inc()

	return fmt.Errorf("failed to push data to queue for tenant=%s and queue_name=%s: queue is full", m.tenantID, m.name)
}

func (m *Queue[T]) StartWorkers() {
	for i := 0; i < m.workerCount; i++ {