Function NewProducer

pkg/ingest/writer_client.go:199–239

NewProducer returns a new KafkaProducer. The input prometheus.Registerer must be wrapped with a prefix (the names of metrics registered don't have a prefix).

(client *kgo.Client, maxBufferedBytes int64, reg prometheus.Registerer)

// The input prometheus.Registerer must be wrapped with a prefix (the names of metrics
// registered don't have a prefix).
func NewProducer(client *kgo.Client, maxBufferedBytes int64, reg prometheus.Registerer) *Producer {
	producer := &Producer{
		Client:           client,
		closeOnce:        &sync.Once{},
		closed:           make(chan struct{}),
		bufferedBytes:    atomic.NewInt64(0),
		maxBufferedBytes: maxBufferedBytes,

		// Metrics.
		bufferedProduceBytes: promauto.With(reg).NewSummary(
			prometheus.SummaryOpts{
				Namespace:  "tempo",
				Subsystem:  "distributor",
				Name:       "buffered_produce_bytes",
				Help:       "The buffered produce records in bytes. Quantile buckets keep track of buffered records size over the last 60s.",
				Objectives: map[float64]float64{0.5: 0.05, 0.99: 0.001, 1: 0.001},
				MaxAge:     time.Minute,
				AgeBuckets: 6,
			}),
		bufferedProduceBytesLimit: promauto.With(reg).NewGauge(
			prometheus.GaugeOpts{
				Namespace: "tempo",
				Subsystem: "distributor",
				Name:      "buffered_produce_bytes_limit",
				Help:      "The bytes limit on buffered produce records. Produce requests fail once this limit is reached.",
			}),

		produceRecordsFailuresTotal: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
			Namespace: "tempo",
			Subsystem: "distributor",
			Name:      "produce_failures_total",
			Help:      "Total number of failed produce records issued to Kafka.",
		}, []string{"reason"}),
	}

	producer.bufferedProduceBytesLimit.Set(float64(maxBufferedBytes))

	go producer.updateMetricsLoop()

	return producer
}

func (c *Producer) Close() {
	c.closeOnce.Do(func() {