Function ExportPartitionLagMetrics

pkg/ingest/metrics.go:47–95

ExportPartitionLagMetrics in a background goroutine by periodically querying Kafka state for the assigned and active partitions. This exports the lag metric in number of records which is different than the lag metric for age. Call ResetLagMetricsForRevokedPartitions when partitions are revoked to p

(ctx context.Context, kclient *kgo.Client, log log.Logger, cfg Config, getAssignedActivePartitions func() []int32, forceMetadataRefresh func())

// stale data. For efficiency this is not detected automatically from changes inthe assigned
// partition callback.
func ExportPartitionLagMetrics(ctx context.Context, kclient *kgo.Client, log log.Logger, cfg Config, getAssignedActivePartitions func() []int32, forceMetadataRefresh func()) {
	go func() {
		var (
			waitTime = cfg.Kafka.ConsumerGroupLagMetricUpdateInterval
			topic    = cfg.Kafka.Topic
			group    = cfg.Kafka.ConsumerGroup
			boff     = backoff.New(ctx, backoff.Config{
				MinBackoff: 100 * time.Millisecond,
				MaxBackoff: waitTime,
				MaxRetries: 5,
			})
			admClient       = kadm.NewClient(kclient)
			partitionClient = NewPartitionOffsetClient(kclient, topic)
		)

		for {
			select {
			case <-time.After(waitTime):
				var (
					lag kadm.GroupLag
					err error
				)
				assignedPartitions := getAssignedActivePartitions()
				boff.Reset()
				for boff.Ongoing() {
					lag, err = getGroupLag(ctx, admClient, partitionClient, group, assignedPartitions)
					if err == nil {
						break
					}
					HandleKafkaError(err, forceMetadataRefresh)
					boff.Wait()
				}

				if err != nil {
					level.Error(log).Log("msg", "metric lag failed:", "err", err, "retries", boff.NumRetries())
					continue
				}
				for _, p := range assignedPartitions {
					l, ok := lag.Lookup(topic, p)
					if ok {
						metricPartitionLag.WithLabelValues(group, strconv.Itoa(int(p))).Set(float64(l.Lag))
					}
				}
			case <-ctx.Done():
				return
			}
		}
	}()
}

// SetPartitionLagSeconds is similar to the auto exported lag, except it is in real clock seconds
// which can only be known after the record is read from the queue, therefore it is set by the caller.