@@ -22,6 +22,7 @@ import (
2222type consumerMetrics struct {
2323 consumeLatency prometheus.Histogram
2424 currentOffset prometheus.Gauge
25+ pushLatency prometheus.Histogram
2526}
2627
2728// newConsumerMetrics initializes and returns a new consumerMetrics instance
@@ -36,6 +37,11 @@ func newConsumerMetrics(reg prometheus.Registerer) *consumerMetrics {
3637 Name : "loki_ingester_partition_current_offset" ,
3738 Help : "The current offset of the Kafka ingester consumer." ,
3839 }),
40+ pushLatency : promauto .With (reg ).NewHistogram (prometheus.HistogramOpts {
41+ Name : "loki_ingester_partition_push_latency_seconds" ,
42+ Help : "The latency of a push request after consuming from Kafka" ,
43+ NativeHistogramBucketFactor : 1.1 ,
44+ }),
3945 }
4046}
4147
@@ -114,7 +120,12 @@ func (kc *kafkaConsumer) consume(ctx context.Context, records []partition.Record
114120 Streams : []logproto.Stream {stream },
115121 }
116122 if err := retryWithBackoff (ctx , func (attempts int ) error {
117- if _ , err := kc .pusher .Push (recordCtx , req ); err != nil {
123+ pushTime := time .Now ()
124+ _ , err := kc .pusher .Push (recordCtx , req )
125+
126+ kc .metrics .pushLatency .Observe (time .Since (pushTime ).Seconds ())
127+
128+ if err != nil {
118129 level .Warn (kc .logger ).Log ("msg" , "failed to push records" , "err" , err , "offset" , record .Offset , "attempts" , attempts )
119130 return err
120131 }
0 commit comments