feat: expose error metric when fetching message (#132)

* feat: expose error metric when fetching message

* chore: fix lint

Author: A.Samet İleri

README.md

@@ -280,7 +280,8 @@ Kafka Konsumer offers an API that handles exposing several metrics.
 
 ### Exposed Metrics
 
-| Metric Name                                       | Description                           | Value Type |
-|---------------------------------------------------|---------------------------------------|------------|
-| kafka_konsumer_processed_messages_total_current   | Total number of processed messages.   | Counter    |
-| kafka_konsumer_unprocessed_messages_total_current | Total number of unprocessed messages. | Counter    |
+| Metric Name                                                      | Description                                    | Value Type |
+|------------------------------------------------------------------|------------------------------------------------|------------|
+| kafka_konsumer_processed_messages_total_current                  | Total number of processed messages.            | Counter    |
+| kafka_konsumer_unprocessed_messages_total_current                | Total number of unprocessed messages.          | Counter    |
+| kafka_konsumer_error_count_during_fetching_message_total_current | Total number of error during fetching message. | Counter    |

collector.go

@@ -11,8 +11,9 @@ const Name = "kafka_konsumer"
 type MetricCollector struct {
 	consumerMetric *ConsumerMetric
 
-	totalUnprocessedMessagesCounter *prometheus.Desc
-	totalProcessedMessagesCounter   *prometheus.Desc
+	totalUnprocessedMessagesCounter      *prometheus.Desc
+	totalProcessedMessagesCounter        *prometheus.Desc
+	totalErrorCountDuringFetchingMessage *prometheus.Desc
 }
 
 func NewMetricCollector(metricPrefix string, consumerMetric *ConsumerMetric) *MetricCollector {
@@ -35,6 +36,12 @@ func NewMetricCollector(metricPrefix string, consumerMetric *ConsumerMetric) *Me
 			emptyStringList,
 			nil,
 		),
+		totalErrorCountDuringFetchingMessage: prometheus.NewDesc(
+			prometheus.BuildFQName(metricPrefix, "error_count_during_fetching_message_total", "current"),
+			"Total number of error during fetching message.",
+			emptyStringList,
+			nil,
+		),
 	}
 }
 
@@ -58,6 +65,13 @@ func (s *MetricCollector) Collect(ch chan<- prometheus.Metric) {
 		float64(s.consumerMetric.TotalUnprocessedMessagesCounter),
 		emptyStringList...,
 	)
+
+	ch <- prometheus.MustNewConstMetric(
+		s.totalErrorCountDuringFetchingMessage,
+		prometheus.CounterValue,
+		float64(s.consumerMetric.TotalErrorCountDuringFetchingMessage),
+		emptyStringList...,
+	)
 }
 
 func NewMetricMiddleware(cfg *ConsumerConfig,

consumer_base.go

@@ -199,6 +199,8 @@ func (c *base) startConsume() {
 				if c.context.Err() != nil {
 					continue
 				}
+
+				c.metric.TotalErrorCountDuringFetchingMessage++
 				c.logger.Warnf("Message could not read, err %s", err.Error())
 				continue
 			}

consumer_base_test.go

@@ -22,6 +22,7 @@ func Test_base_startConsume(t *testing.T) {
 			pause:                 make(chan struct{}),
 			logger:                NewZapLogger(LogLevelError),
 			consumerState:         stateRunning,
+			metric:                &ConsumerMetric{},
 		}
 		b.context, b.cancelFn = context.WithCancel(context.Background())
 

metric.go

@@ -1,6 +1,7 @@
 package kafka
 
 type ConsumerMetric struct {
-	TotalUnprocessedMessagesCounter int64
-	TotalProcessedMessagesCounter   int64
+	TotalUnprocessedMessagesCounter      int64
+	TotalProcessedMessagesCounter        int64
+	TotalErrorCountDuringFetchingMessage int64
 }