feat: adding metric prefix config when multiple prometheus metric reg…

batch_consumer.go

@@ -3,6 +3,8 @@
 import (
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/segmentio/kafka-go"
 
 	kcronsumer "github.com/Trendyol/kafka-cronsumer/pkg/kafka"
@@ -43,6 +45,10 @@
 	return &c, nil
 }
 
+func (b *batchConsumer) GetMetricCollectors() []prometheus.Collector {
+	return b.base.GetMetricCollectors()
+}
+
 func (b *batchConsumer) GetMetric() *ConsumerMetric {
 	return b.metric
 }

collector.go

@@ -8,20 +8,20 @@
 
 const Name = "kafka_konsumer"
 
-type metricCollector struct {
+type MetricCollector struct {
 	consumerMetric *ConsumerMetric
 
 	totalUnprocessedMessagesCounter *prometheus.Desc
 	totalProcessedMessagesCounter   *prometheus.Desc
 }
 
-func (s *metricCollector) Describe(ch chan<- *prometheus.Desc) {
+func (s *MetricCollector) Describe(ch chan<- *prometheus.Desc) {
 	prometheus.DescribeByCollect(s, ch)
 }
 
 var emptyStringList []string
 
-func (s *metricCollector) Collect(ch chan<- prometheus.Metric) {
+func (s *MetricCollector) Collect(ch chan<- prometheus.Metric) {
 	ch <- prometheus.MustNewConstMetric(
 		s.totalProcessedMessagesCounter,
 		prometheus.CounterValue,
@@ -37,18 +37,22 @@
 	)
 }
 
-func newMetricCollector(consumerMetric *ConsumerMetric) *metricCollector {
-	return &metricCollector{
+func NewMetricCollector(consumerMetric *ConsumerMetric, prefix string) *MetricCollector {
+	if prefix == "" {
+		prefix = Name
+	}
+
+	return &MetricCollector{
 		consumerMetric: consumerMetric,
 
 		totalProcessedMessagesCounter: prometheus.NewDesc(
-			prometheus.BuildFQName(Name, "processed_messages_total", "current"),
+			prometheus.BuildFQName(prefix, "processed_messages_total", "current"),
 			"Total number of processed messages.",
 			emptyStringList,
 			nil,
 		),
 		totalUnprocessedMessagesCounter: prometheus.NewDesc(
-			prometheus.BuildFQName(Name, "unprocessed_messages_total", "current"),
+			prometheus.BuildFQName(prefix, "unprocessed_messages_total", "current"),
 			"Total number of unprocessed messages.",
 			emptyStringList,
 			nil,
@@ -61,7 +65,7 @@
 	consumerMetric *ConsumerMetric,
 	metricCollectors ...prometheus.Collector,
 ) (func(ctx *fiber.Ctx) error, error) {
-	prometheus.DefaultRegisterer.MustRegister(newMetricCollector(consumerMetric))
+	prometheus.DefaultRegisterer.MustRegister(NewMetricCollector(consumerMetric, cfg.MetricPrefix))
 	prometheus.DefaultRegisterer.MustRegister(metricCollectors...)
 
 	fiberPrometheus := fiberprometheus.New(cfg.Reader.GroupID)

collector_test.go

@@ -0,0 +1,60 @@
+package kafka
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+func Test_NewCollector(t *testing.T) {
+	t.Run("When_Default_Prefix_Value_Used", func(t *testing.T) {
+		cronsumerMetric := &ConsumerMetric{}
+		expectedTotalProcessedMessagesCounter := prometheus.NewDesc(
+			prometheus.BuildFQName(Name, "processed_messages_total", "current"),
+			"Total number of processed messages.",
+			emptyStringList,
+			nil,
+		)
+		expectedTotalUnprocessedMessagesCounter := prometheus.NewDesc(
+			prometheus.BuildFQName(Name, "unprocessed_messages_total", "current"),
+			"Total number of unprocessed messages.",
+			emptyStringList,
+			nil,
+		)
+
+		collector := NewMetricCollector(cronsumerMetric, "")
+
+		if !reflect.DeepEqual(collector.totalProcessedMessagesCounter, expectedTotalProcessedMessagesCounter) {
+			t.Errorf("Expected: %+v, Actual: %+v", collector.totalProcessedMessagesCounter, expectedTotalProcessedMessagesCounter)
+		}
+		if !reflect.DeepEqual(collector.totalUnprocessedMessagesCounter, expectedTotalUnprocessedMessagesCounter) {
+			t.Errorf("Expected: %+v, Actual: %+v", collector.totalUnprocessedMessagesCounter, expectedTotalUnprocessedMessagesCounter)
+		}
+	})
+	t.Run("When_Custom_Prefix_Value_Used", func(t *testing.T) {
+		cronsumerMetric := &ConsumerMetric{}
+		customPrefix := "custom_prefix"
+		expectedTotalProcessedMessagesCounter := prometheus.NewDesc(
+			prometheus.BuildFQName(customPrefix, "processed_messages_total", "current"),
+			"Total number of processed messages.",
+			emptyStringList,
+			nil,
+		)
+		expectedTotalUnprocessedMessagesCounter := prometheus.NewDesc(
+			prometheus.BuildFQName(customPrefix, "unprocessed_messages_total", "current"),
+			"Total number of unprocessed messages.",
+			emptyStringList,
+			nil,
+		)
+
+		collector := NewMetricCollector(cronsumerMetric, customPrefix)
+
+		if !reflect.DeepEqual(collector.totalProcessedMessagesCounter, expectedTotalProcessedMessagesCounter) {
+			t.Errorf("Expected: %+v, Actual: %+v", collector.totalProcessedMessagesCounter, expectedTotalProcessedMessagesCounter)
+		}
+		if !reflect.DeepEqual(collector.totalUnprocessedMessagesCounter, expectedTotalUnprocessedMessagesCounter) {
+			t.Errorf("Expected: %+v, Actual: %+v", collector.totalUnprocessedMessagesCounter, expectedTotalUnprocessedMessagesCounter)
+		}
+	})
+}

consumer.go

@@ -3,6 +3,8 @@
 import (
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/segmentio/kafka-go"
 
 	kcronsumer "github.com/Trendyol/kafka-cronsumer/pkg/kafka"
@@ -49,6 +51,10 @@
 	go c.startBatch()
 }
 
+func (c *consumer) GetMetricCollectors() []prometheus.Collector {
+	return c.base.GetMetricCollectors()
+}
+
 func (c *consumer) startBatch() {
 	defer c.wg.Done()
 

consumer_base.go

@@ -25,6 +25,11 @@
 
 	// Stop for graceful shutdown. In order to avoid data loss, you have to call it!
 	Stop() error
+
+	// GetMetricCollectors for the purpose of making metric collectors available.
+	// You can register these collectors on your own http server.
+	// Please look at the examples/with-metric-collector directory.
+	GetMetricCollectors() []prometheus.Collector
 }
 
 type Reader interface {
@@ -56,6 +61,7 @@
 	retryEnabled              bool
 	transactionalRetry        bool
 	distributedTracingEnabled bool
+	metricSuffix              string
 }
 
 func NewConsumer(cfg *ConsumerConfig) (Consumer, error) {
@@ -90,6 +96,7 @@
 		messageProcessedStream:    make(chan struct{}, cfg.Concurrency),
 		singleConsumingStream:     make(chan *Message, cfg.Concurrency),
 		batchConsumingStream:      make(chan []*Message, cfg.Concurrency),
+		metricSuffix:              cfg.MetricPrefix,
 	}
 
 	if cfg.DistributedTracingEnabled {
@@ -120,6 +127,17 @@
 	c.subprocesses.Add(c.api)
 }
 
+func (c *base) GetMetricCollectors() []prometheus.Collector {
+	var metricCollectors []prometheus.Collector
+	if c.retryEnabled {
+		metricCollectors = c.cronsumer.GetMetricCollectors()
+	}
+
+	metricCollectors = append(metricCollectors, NewMetricCollector(c.metric, c.metricSuffix))
+
+	return metricCollectors
+}
+
 func (c *base) startConsume() {
 	defer c.wg.Done()
 

consumer_config.go

@@ -48,12 +48,20 @@
 	DistributedTracingEnabled       bool
 	RetryEnabled                    bool
 	APIEnabled                      bool
+
+	// MetricPrefix is used for prometheus fq name prefix.
+	// If not provided, default metric prefix value is `kafka_konsumer`.
+	// Currently, there are two exposed prometheus metrics. `processed_messages_total` and `unprocessed_messages_total`.
+	// So, if default metric prefix used, metrics names are `kafka_konsumer_processed_messages_total_current` and
+	// `kafka_konsumer_unprocessed_messages_total_current`.
+	MetricPrefix string
 }
 
 func (cfg *ConsumerConfig) newCronsumerConfig() *kcronsumer.Config {
 	cronsumerCfg := kcronsumer.Config{
-		ClientID: cfg.RetryConfiguration.ClientID,
-		Brokers:  cfg.RetryConfiguration.Brokers,
+		MetricPrefix: cfg.RetryConfiguration.MetricPrefix,
+		ClientID:     cfg.RetryConfiguration.ClientID,
+		Brokers:      cfg.RetryConfiguration.Brokers,
 		Consumer: kcronsumer.ConsumerConfig{
 			ClientID:          cfg.ClientID,
 			GroupID:           cfg.Reader.GroupID,
@@ -122,6 +130,7 @@
 	Brokers         []string
 	MaxRetry        int
 	WorkDuration    time.Duration
+	MetricPrefix    string
 }
 
 type BatchConfiguration struct {

examples/docker-compose.yml

@@ -49,6 +49,7 @@ services:
     command: "bash -c 'echo Waiting for Kafka to be ready... && \
                          cub kafka-ready -b kafka:9092 1 20 && \
                          kafka-topics --create --topic standart-topic --if-not-exists --zookeeper zookeeper:2181 --partitions 1 --replication-factor 1 && \
+                         kafka-topics --create --topic another-standart-topic --if-not-exists --zookeeper zookeeper:2181 --partitions 1 --replication-factor 1 && \
                          kafka-topics --create --topic retry-topic --if-not-exists --zookeeper zookeeper:2181 --partitions 1 --replication-factor 1 && \
                          kafka-topics --create --topic error-topic --if-not-exists --zookeeper zookeeper:2181 --partitions 1 --replication-factor 1 && \
                          sleep infinity'"

examples/with-metric-collector/README.md

@@ -0,0 +1,37 @@
+If you run this example and go to http://localhost:8000/metrics, 
+
+you can see first and second consumet metrics as shown below
+
+```
+# HELP first_discarded_messages_total_current Total number of discarded messages.
+# TYPE first_discarded_messages_total_current counter
+first_discarded_messages_total_current 0
+
+# HELP first_processed_messages_total_current Total number of processed messages.
+# TYPE first_processed_messages_total_current counter
+first_processed_messages_total_current 0
+
+# HELP first_retried_messages_total_current Total number of retried messages.
+# TYPE first_retried_messages_total_current counter
+first_retried_messages_total_current 0
+
+# HELP first_unprocessed_messages_total_current Total number of unprocessed messages.
+# TYPE first_unprocessed_messages_total_current counter
+first_unprocessed_messages_total_current 0
+
+# HELP second_discarded_messages_total_current Total number of discarded messages.
+# TYPE second_discarded_messages_total_current counter
+second_discarded_messages_total_current 0
+
+# HELP second_processed_messages_total_current Total number of processed messages.
+# TYPE second_processed_messages_total_current counter
+second_processed_messages_total_current 0
+
+# HELP second_retried_messages_total_current Total number of retried messages.
+# TYPE second_retried_messages_total_current counter
+second_retried_messages_total_current 0
+
+# HELP second_unprocessed_messages_total_current Total number of unprocessed messages.
+# TYPE second_unprocessed_messages_total_current counter
+second_unprocessed_messages_total_current 0
+```

examples/with-metric-collector/api.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"fmt"
+	"github.com/gofiber/fiber/v2"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const port = 8000
+
+func StartAPI(metricCollectors ...prometheus.Collector) {
+	f := fiber.New(
+		fiber.Config{
+			DisableStartupMessage:    true,
+			DisableDefaultDate:       true,
+			DisableHeaderNormalizing: true,
+		},
+	)
+
+	metricMiddleware, err := NewMetricMiddleware(f, metricCollectors...)
+
+	if err == nil {
+		f.Use(metricMiddleware)
+	} else {
+		fmt.Printf("metric middleware cannot be initialized: %v", err)
+	}
+
+	fmt.Printf("server starting on port %d", port)
+
+	go listen(f)
+}
+
+func listen(f *fiber.App) {
+	if err := f.Listen(fmt.Sprintf(":%d", port)); err != nil {
+		fmt.Printf("server cannot start on port %d, err: %v", port, err)
+	}
+}

examples/with-metric-collector/main.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"fmt"
+	"github.com/Trendyol/kafka-konsumer/v2"
+	"time"
+)
+
+func main() {
+	firstConsumerCfg := &kafka.ConsumerConfig{
+		MetricPrefix: "first",
+		Reader: kafka.ReaderConfig{
+			Brokers: []string{"localhost:29092"},
+			Topic:   "standart-topic",
+			GroupID: "standart-cg",
+		},
+		RetryEnabled: true,
+		RetryConfiguration: kafka.RetryConfiguration{
+			MetricPrefix:  "first",
+			Brokers:       []string{"localhost:29092"},
+			Topic:         "error-topic",
+			StartTimeCron: "*/1 * * * *",
+			WorkDuration:  50 * time.Second,
+			MaxRetry:      3,
+		},
+		ConsumeFn: consumeFn,
+	}
+
+	firstConsumer, _ := kafka.NewConsumer(firstConsumerCfg)
+	defer firstConsumer.Stop()
+
+	go firstConsumer.Consume()
+
+	secondConsumerCfg := &kafka.ConsumerConfig{
+		MetricPrefix: "second",
+		Reader: kafka.ReaderConfig{
+			Brokers: []string{"localhost:29092"},
+			Topic:   "another-standart-topic",
+			GroupID: "another-standart-cg",
+		},
+		RetryEnabled: true,
+		RetryConfiguration: kafka.RetryConfiguration{
+			MetricPrefix:  "second",
+			Brokers:       []string{"localhost:29092"},
+			Topic:         "retry-topic",
+			StartTimeCron: "*/1 * * * *",
+			WorkDuration:  50 * time.Second,
+			MaxRetry:      3,
+		},
+		ConsumeFn: consumeFn,
+	}
+
+	secondConsumer, _ := kafka.NewConsumer(secondConsumerCfg)
+	defer secondConsumer.Stop()
+
+	go secondConsumer.Consume()
+
+	allCollectors := append(firstConsumer.GetMetricCollectors(), secondConsumer.GetMetricCollectors()...)
+	StartAPI(allCollectors...)
+
+	select {}
+}
+
+func consumeFn(message *kafka.Message) error {
+	fmt.Printf("Message From %s with value %s", message.Topic, string(message.Value))
+	return nil
+}