Add metric fission_mqt_message_lag for kafka mqt connector (#2544)

These changes have specifically been made for Kafka connector. This will expose a new metric named `fission_mqt_message_lag`, which will show a number of messages lag per topic and partition.

We can use this metric in the auto-scaling of the pod for the new deploy type executor function. While creating a new deploy function we need to add hpa metrics of external type inside the function definition.

pkg/mqtrigger/messageQueue/kafka/consumer.go

@@ -100,6 +100,15 @@ func (ch MqtConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) e
 
 // ConsumeClaims implemented to satisfy the sarama.ConsumerGroupHandler interface
 func (ch MqtConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+
+	trigger := ch.trigger.Name
+	triggerNamespace := ch.trigger.Namespace
+	topic := claim.Topic()
+	partition := string(claim.Partition())
+
+	// initially set metrics to -1
+	mqtrigger.SetMessageLagCount(trigger, triggerNamespace, topic, partition, -1)
+
 	// Do not move the code below to a goroutine.
 	// The `ConsumeClaim` itself is called within a goroutine
 	for {
@@ -108,8 +117,12 @@ func (ch MqtConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSessi
 			if msg != nil {
 				ch.kafkaMsgHandler(msg)
 				session.MarkMessage(msg, "")
-				mqtrigger.IncreaseMessageCount(ch.trigger.Name, ch.trigger.Namespace)
+				mqtrigger.IncreaseMessageCount(trigger, triggerNamespace)
 			}
+
+			mqtrigger.SetMessageLagCount(trigger, triggerNamespace, topic, partition,
+				claim.HighWaterMarkOffset()-msg.Offset-1)
+
 		// Should return when `session.Context()` is done.
 		case <-session.Context().Done():
 			return nil

pkg/mqtrigger/metrics.go

@@ -37,6 +37,13 @@ var (
 		},
 		labels,
 	)
+	messageLagCount = promauto.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "fission_mqt_message_lag",
+			Help: "Total number of messages lag per topic and partition",
+		},
+		[]string{"trigger_name", "trigger_namespace", "topic", "partition"},
+	)
 )
 
 func IncreaseSubscriptionCount() {
@@ -50,3 +57,7 @@ func DecreaseSubscriptionCount() {
 func IncreaseMessageCount(trigname, trignamespace string) {
 	messageCount.WithLabelValues(trigname, trignamespace).Inc()
 }
+
+func SetMessageLagCount(trigname, trignamespace, topic, partition string, lag int64) {
+	messageLagCount.WithLabelValues(trigname, trignamespace, topic, partition).Set(float64(lag))
+}