Add internal metrics

Author: weeco

kafka/consumer_group_metadata.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	log "github.com/sirupsen/logrus"
+	"strconv"
 )
 
 type consumerGroupMetadata struct {
@@ -53,6 +54,7 @@ func newConsumerGroupMetadata(key *bytes.Buffer, value *bytes.Buffer, logger *lo
 
 		return nil, err
 	}
+	groupMetadata.WithLabelValues(strconv.Itoa(int(valueVersion))).Add(1)
 
 	// Decode value content
 	var metadata *consumerGroupMetadata

kafka/consumer_partition_offset.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	log "github.com/sirupsen/logrus"
+	"strconv"
 )
 
 // ConsumerPartitionOffset represents a consumer group commit which can be decoded from the consumer_offsets topic
@@ -69,6 +70,7 @@ func newConsumerPartitionOffset(key *bytes.Buffer, value *bytes.Buffer, logger *
 
 		return nil, fmt.Errorf("message value has no version")
 	}
+	offsetCommit.WithLabelValues(strconv.Itoa(int(valueVersion))).Add(1)
 
 	// Decode message value using the right decoding function for given version
 	var decodedValue offsetValue

kafka/metrics.go

@@ -0,0 +1,47 @@
+package kafka
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const internalMetricsName = "kafka_minion_internal"
+
+var (
+	offsetCommit = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: prometheus.BuildFQName(internalMetricsName, "offset_consumer", "offset_commits_read"),
+		Help: "Number of read offset commits",
+	}, []string{"version"})
+	offsetCommitTombstone = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: prometheus.BuildFQName(internalMetricsName, "offset_consumer", "offset_commits_tombstones_read"),
+		Help: "Number of read group offset commit tombstone messages",
+	})
+
+	groupMetadata = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: prometheus.BuildFQName(internalMetricsName, "offset_consumer", "group_metadata_read"),
+		Help: "Number of read group meta data messages",
+	}, []string{"version"})
+	groupMetadataTombstone = prometheus.NewCounter(prometheus.CounterOpts{
+		Name: prometheus.BuildFQName(internalMetricsName, "offset_consumer", "group_metadata_tombstones_read"),
+		Help: "Number of read group meta data tombstone messages",
+	})
+
+	messagesInSuccess = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: prometheus.BuildFQName(internalMetricsName, "kafka", "messages_in_success"),
+		Help: "Number of messages successfully consumed from a topic",
+	}, []string{"topic"})
+	messagesInFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: prometheus.BuildFQName(internalMetricsName, "kafka", "messages_in_failed"),
+		Help: "Number of messages failed to consume from a topic",
+	}, []string{"topic"})
+)
+
+func init() {
+	prometheus.MustRegister(offsetCommit)
+	prometheus.MustRegister(offsetCommitTombstone)
+
+	prometheus.MustRegister(groupMetadata)
+	prometheus.MustRegister(groupMetadataTombstone)
+
+	prometheus.MustRegister(messagesInSuccess)
+	prometheus.MustRegister(messagesInFailed)
+}

kafka/offset_consumer.go

@@ -108,8 +108,10 @@ func (module *OffsetConsumer) partitionConsumer(consumer sarama.Consumer, partit
 	for {
 		select {
 		case msg := <-pconsumer.Messages():
+			messagesInSuccess.WithLabelValues(msg.Topic).Add(1)
 			module.processMessage(msg)
 		case err := <-pconsumer.Errors():
+			messagesInFailed.WithLabelValues(err.Topic).Add(1)
 			log.WithFields(log.Fields{
 				"error":     err.Error(),
 				"topic":     err.Topic,
@@ -162,6 +164,7 @@ func (module *OffsetConsumer) processOffsetCommit(key *bytes.Buffer, value *byte
 	// A tombstone on the __consumer_offsets topic indicates that the consumer group either expired
 	// due too configured group retention or that the consumed topic has been deleted
 	if isTombstone {
+		offsetCommitTombstone.Add(1)
 		group, err := readString(key)
 		if err != nil {
 			logger.WithFields(log.Fields{