Only expose metrics if topic has been consumed

Author: Martin Schneppenheim

collector/collector.go

@@ -108,6 +108,12 @@ func (e *Collector) Describe(ch chan<- *prometheus.Desc) {
 // Collect is triggered by the Prometheus registry when the metrics endpoint has been invoked
 func (e *Collector) Collect(ch chan<- prometheus.Metric) {
 	log.Debug("Collector's collect has been invoked")
+
+	if e.storage.IsConsumed() == false {
+		log.Info("Offets topic has not yet been consumed until the end")
+		return
+	}
+
 	consumerOffsets := e.storage.ConsumerOffsets()
 	partitionLowWaterMarks := e.storage.PartitionLowWaterMarks()
 	partitionHighWaterMarks := e.storage.PartitionHighWaterMarks()

kafka/cluster.go

@@ -29,6 +29,23 @@ type PartitionWaterMark struct {
 	Timestamp   int64
 }
 
+type consumerOffsetTopic struct {
+	Lock           sync.RWMutex
+	PartitionsByID map[int32]consumerOffsetPartition
+}
+
+type consumerOffsetPartition struct {
+	PartitionID   int32
+	HighWaterMark int64
+}
+
+var (
+	offsetWaterMarks = consumerOffsetTopic{
+		Lock:           sync.RWMutex{},
+		PartitionsByID: make(map[int32]consumerOffsetPartition),
+	}
+)
+
 // NewCluster creates a new cluster module and tries to connect to the kafka cluster
 // If it cannot connect to the cluster it will panic
 func NewCluster(opts *options.Options, partitionWaterMarksCh chan *StorageRequest) *Cluster {
@@ -190,9 +207,6 @@ func (module *Cluster) getHighWaterMarks(wg *sync.WaitGroup, broker *sarama.Brok
 	}
 	ts := time.Now().Unix() * 1000
 	for topicName, responseBlock := range response.Blocks {
-		if !module.isTopicAllowed(topicName) {
-			continue
-		}
 
 		for partitionID, offsetResponse := range responseBlock {
 			if offsetResponse.Err != sarama.ErrNoError {
@@ -204,6 +218,19 @@ func (module *Cluster) getHighWaterMarks(wg *sync.WaitGroup, broker *sarama.Brok
 				continue
 			}
 
+			if topicName == module.options.ConsumerOffsetsTopicName {
+				offsetWaterMarks.Lock.Lock()
+				offsetWaterMarks.PartitionsByID[partitionID] = consumerOffsetPartition{
+					PartitionID:   partitionID,
+					HighWaterMark: offsetResponse.Offsets[0],
+				}
+				offsetWaterMarks.Lock.Unlock()
+			}
+
+			// Skip topic in this for loop (instead of the outer one) because we still need __consumer_offset information
+			if !module.isTopicAllowed(topicName) {
+				continue
+			}
 			logger.WithFields(log.Fields{
 				"topic":     topicName,
 				"partition": partitionID,

kafka/offset_consumer.go

@@ -6,10 +6,17 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/google-cloud-tools/kafka-minion/options"
 	log "github.com/sirupsen/logrus"
+	"math"
 	"strings"
 	"sync"
+	"time"
 )
 
+type consumerStatus struct {
+	PartitionLag int64
+	IsReady      bool // Indicates whether a partition consumer has caught up the partition lag or not
+}
+
 // OffsetConsumer is a consumer module which reads consumer group information from the offsets topic in a Kafka cluster.
 // The offsets topic is typically named __consumer_offsets. All messages in this topic are binary and therefore they
 // must first be decoded to access the information. This module consumes and processes all messages in the offsets topic.
@@ -81,6 +88,8 @@ func (module *OffsetConsumer) Start() {
 	}).Infof("Starting '%d' partition consumers", len(partitions))
 	for _, partition := range partitions {
 		module.wg.Add(1)
+		registerPartitionRequest := newRegisterOffsetPartition(partition)
+		module.storageChannel <- registerPartitionRequest
 		go module.partitionConsumer(consumer, partition)
 	}
 	log.WithFields(log.Fields{
@@ -105,18 +114,38 @@ func (module *OffsetConsumer) partitionConsumer(consumer sarama.Consumer, partit
 	log.Debugf("Started consumer %d", partitionID)
 	defer pconsumer.AsyncClose()
 
+	ticker := time.NewTicker(5 * time.Second)
+	var consumedOffset int64
+
 	for {
 		select {
 		case msg := <-pconsumer.Messages():
 			messagesInSuccess.WithLabelValues(msg.Topic).Add(1)
 			module.processMessage(msg)
+			consumedOffset = msg.Offset
 		case err := <-pconsumer.Errors():
 			messagesInFailed.WithLabelValues(err.Topic).Add(1)
 			log.WithFields(log.Fields{
 				"error":     err.Error(),
 				"topic":     err.Topic,
 				"partition": err.Partition,
 			}).Errorf("partition consume error")
+		case <-ticker.C:
+			var highWaterMark int64
+			highWaterMark = math.MaxInt64
+			offsetWaterMarks.Lock.RLock()
+			if val, exists := offsetWaterMarks.PartitionsByID[partitionID]; exists {
+				// Not sure why -1 is needed here, but otherwise there are lots of partition consumers with a remaining lag of 1
+				highWaterMark = val.HighWaterMark - 1
+			}
+			offsetWaterMarks.Lock.RUnlock()
+			if consumedOffset >= highWaterMark {
+				request := newMarkOffsetPartitionReady(partitionID)
+				module.storageChannel <- request
+				ticker.Stop()
+			} else {
+				log.Debugf("Not ready, Lag is: %v (consumed: %v)", highWaterMark-consumedOffset, consumedOffset)
+			}
 		}
 	}
 }

kafka/storage_request.go

@@ -15,6 +15,14 @@ const (
 
 	// StorageDeleteConsumerGroup is the request type to remove an offset commit for a topic:group:partition combination
 	StorageDeleteConsumerGroup StorageRequestType = 3
+
+	// StorageRegisterOffsetPartition is the request type to make the storage module aware that a partition consumer for
+	// the consumer offsets partition exists and that it should await it's ready signal before exposing metrics
+	StorageRegisterOffsetPartition StorageRequestType = 4
+
+	// StorageMarkOffsetPartitionReady is the request type to mark a partition consumer of the consumer offsets topic
+	// as ready (=caught up partition lag)
+	StorageMarkOffsetPartitionReady StorageRequestType = 5
 )
 
 // StorageRequest is an entity to send requests to the storage module
@@ -56,3 +64,17 @@ func newDeleteConsumerGroupRequest(group string, topic string, partitionID int32
 		PartitionID:       partitionID,
 	}
 }
+
+func newRegisterOffsetPartition(partitionID int32) *StorageRequest {
+	return &StorageRequest{
+		RequestType: StorageRegisterOffsetPartition,
+		PartitionID: partitionID,
+	}
+}
+
+func newMarkOffsetPartitionReady(partitionID int32) *StorageRequest {
+	return &StorageRequest{
+		RequestType: StorageMarkOffsetPartitionReady,
+		PartitionID: partitionID,
+	}
+}

storage/offset.go

@@ -13,6 +13,10 @@ type OffsetStorage struct {
 	consumerOffsetCh chan *kafka.StorageRequest
 	clusterCh        chan *kafka.StorageRequest
 
+	notReadyPartitionConsumers int32
+	offsetTopicConsumed        bool
+
+	consumerStatusLock          sync.RWMutex // Lock is being used if you either access notReadyPartitionConsumers or offsetTopicConsumed
 	consumerOffsetsLock         sync.RWMutex
 	partitionHighWaterMarksLock sync.RWMutex
 	partitionLowWaterMarksLock  sync.RWMutex
@@ -22,6 +26,8 @@ type OffsetStorage struct {
 	consumerOffsets         map[string]ConsumerPartitionOffsetMetric
 	partitionHighWaterMarks map[string]kafka.PartitionWaterMark
 	partitionLowWaterMarks  map[string]kafka.PartitionWaterMark
+
+	logger *log.Entry
 }
 
 // ConsumerPartitionOffsetMetric represents an offset commit but is extended by further fields which can be
@@ -41,13 +47,20 @@ func NewOffsetStorage(consumerOffsetCh chan *kafka.StorageRequest, clusterCh cha
 		consumerOffsetCh: consumerOffsetCh,
 		clusterCh:        clusterCh,
 
+		notReadyPartitionConsumers: 0,
+		offsetTopicConsumed:        false,
+
 		consumerOffsetsLock:         sync.RWMutex{},
 		partitionHighWaterMarksLock: sync.RWMutex{},
 		partitionLowWaterMarksLock:  sync.RWMutex{},
 
 		consumerOffsets:         make(map[string]ConsumerPartitionOffsetMetric),
 		partitionHighWaterMarks: make(map[string]kafka.PartitionWaterMark),
 		partitionLowWaterMarks:  make(map[string]kafka.PartitionWaterMark),
+
+		logger: log.WithFields(log.Fields{
+			"module": "storage",
+		}),
 	}
 }
 
@@ -64,6 +77,11 @@ func (module *OffsetStorage) consumerOffsetWorker() {
 			module.storeOffsetEntry(request.ConsumerOffset)
 		case kafka.StorageDeleteConsumerGroup:
 			module.deleteOffsetEntry(request.ConsumerGroupName, request.TopicName, request.PartitionID)
+		case kafka.StorageRegisterOffsetPartition:
+			module.registerOffsetPartition(request.PartitionID)
+		case kafka.StorageMarkOffsetPartitionReady:
+			module.markOffsetPartitionReady(request.PartitionID)
+
 		default:
 			log.WithFields(log.Fields{
 				"request_type": request.RequestType,
@@ -98,6 +116,24 @@ func (module *OffsetStorage) storePartitionHighWaterMark(offset *kafka.Partition
 	module.partitionHighWaterMarksLock.Unlock()
 }
 
+func (module *OffsetStorage) registerOffsetPartition(partitionID int32) {
+	module.consumerOffsetsLock.Lock()
+	defer module.consumerOffsetsLock.Unlock()
+
+	module.notReadyPartitionConsumers++
+}
+
+func (module *OffsetStorage) markOffsetPartitionReady(partitionID int32) {
+	module.consumerOffsetsLock.Lock()
+	defer module.consumerOffsetsLock.Unlock()
+
+	module.notReadyPartitionConsumers--
+	if module.notReadyPartitionConsumers == 0 {
+		module.logger.Info("Offset topic has been consumed")
+		module.offsetTopicConsumed = true
+	}
+}
+
 func (module *OffsetStorage) storePartitionLowWaterMark(offset *kafka.PartitionWaterMark) {
 	key := fmt.Sprintf("%v:%v", offset.TopicName, offset.PartitionID)
 	module.partitionLowWaterMarksLock.Lock()
@@ -170,3 +206,12 @@ func (module *OffsetStorage) PartitionLowWaterMarks() map[string]kafka.Partition
 
 	return mapCopy
 }
+
+// IsConsumed indicates whether the consumer offsets topic lag has been caught up and therefore
+// the metrics reported by this module are accurate or not
+func (module *OffsetStorage) IsConsumed() bool {
+	module.consumerStatusLock.RLock()
+	defer module.consumerStatusLock.RUnlock()
+
+	return module.offsetTopicConsumed
+}