Fixed mqtrigger scaling issue (#2986)

* Fixed mqtrigger scaling issue

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

* Resolve review comments

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

* Set goreleaser version to v1
Remove armv7 references from goreleaser file

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

---------

Signed-off-by: Md Soharab Ansari <soharab.ansari@infracloud.io>

pkg/mqtrigger/messageQueue/kafka/consumer.go

@@ -74,6 +74,8 @@ func NewMqtConsumerGroupHandler(version sarama.KafkaVersion,
 
 // Setup implemented to satisfy the sarama.ConsumerGroupHandler interface
 func (ch MqtConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
+	mqtrigger.SetTriggerStatus(ch.trigger.ObjectMeta.Name, ch.trigger.ObjectMeta.Namespace)
+	mqtrigger.IncreaseInprocessCount()
 	ch.logger.With(
 		zap.String("trigger", ch.trigger.ObjectMeta.Name),
 		zap.String("topic", ch.trigger.Spec.Topic),
@@ -88,6 +90,8 @@ func (ch MqtConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) err
 
 // Cleanup implemented to satisfy the sarama.ConsumerGroupHandler interface
 func (ch MqtConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
+	mqtrigger.ResetTriggerStatus(ch.trigger.ObjectMeta.Name, ch.trigger.ObjectMeta.Namespace)
+	mqtrigger.DecreaseInprocessCount()
 	ch.logger.With(
 		zap.String("trigger", ch.trigger.ObjectMeta.Name),
 		zap.String("topic", ch.trigger.Spec.Topic),

pkg/mqtrigger/messageQueue/kafka/kafka.go

@@ -54,7 +54,6 @@ type (
 		routerUrl string
 		brokers   []string
 		version   sarama.KafkaVersion
-		client    sarama.Client
 		authKeys  map[string][]byte
 		tls       bool
 	}
@@ -111,18 +110,24 @@ func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messa
 
 	logger.Info("created kafka queue", zap.Any("kafka brokers", kafka.brokers),
 		zap.Any("kafka version", kafka.version))
+	return kafka, nil
+}
 
-	// Create new config
-	saramaConfig := sarama.NewConfig()
-	saramaConfig.Version = kafka.version
+func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error) {
+	kafka.logger.Debug("inside kakfa subscribe", zap.Any("trigger", trigger))
+	kafka.logger.Debug("brokers set", zap.Strings("brokers", kafka.brokers))
 
-	// consumer config
-	saramaConfig.Consumer.Return.Errors = true
+	// Create new consumer
+	consumerConfig := sarama.NewConfig()
+	consumerConfig.Consumer.Return.Errors = true
+	consumerConfig.Version = kafka.version
 
-	// producer config
-	saramaConfig.Producer.RequiredAcks = sarama.WaitForAll
-	saramaConfig.Producer.Retry.Max = 10
-	saramaConfig.Producer.Return.Successes = true
+	// Create new producer
+	producerConfig := sarama.NewConfig()
+	producerConfig.Producer.RequiredAcks = sarama.WaitForAll
+	producerConfig.Producer.Retry.Max = 10
+	producerConfig.Producer.Return.Successes = true
+	producerConfig.Version = kafka.version
 
 	// Setup TLS for both producer and consumer
 	if kafka.tls {
@@ -132,30 +137,18 @@ func New(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messa
 			return nil, err
 		}
 
-		saramaConfig.Net.TLS.Enable = true
-		saramaConfig.Net.TLS.Config = tlsConfig
+		producerConfig.Net.TLS.Enable = true
+		producerConfig.Net.TLS.Config = tlsConfig
+		consumerConfig.Net.TLS.Enable = true
+		consumerConfig.Net.TLS.Config = tlsConfig
 	}
 
-	saramaClient, err := sarama.NewClient(kafka.brokers, saramaConfig)
+	consumer, err := sarama.NewConsumerGroup(kafka.brokers, string(trigger.ObjectMeta.UID), consumerConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	kafka.client = saramaClient
-
-	return kafka, nil
-}
-
-func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Subscription, error) {
-	kafka.logger.Debug("inside kakfa subscribe", zap.Any("trigger", trigger))
-	kafka.logger.Debug("brokers set", zap.Strings("brokers", kafka.brokers))
-
-	consumer, err := sarama.NewConsumerGroupFromClient(string(trigger.ObjectMeta.UID), kafka.client)
-	if err != nil {
-		return nil, err
-	}
-
-	producer, err := sarama.NewSyncProducerFromClient(kafka.client)
+	producer, err := sarama.NewSyncProducer(kafka.brokers, producerConfig)
 	if err != nil {
 		return nil, err
 	}

pkg/mqtrigger/metrics.go

@@ -45,6 +45,20 @@ var (
 		},
 		[]string{"trigger_name", "trigger_namespace", "topic", "partition"},
 	)
+	triggerStatus = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "fission_mqt_status",
+			Help: "Status of an individual trigger 1 if processing otherwise 0",
+		},
+		[]string{"trigger_name", "trigger_namespace"},
+	)
+	mqtInprocessCount = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "fission_mqt_inprocess",
+			Help: "Total number of MQTs in active processing",
+		},
+		[]string{},
+	)
 )
 
 func IncreaseSubscriptionCount() {
@@ -55,6 +69,22 @@ func DecreaseSubscriptionCount() {
 	subscriptionCount.WithLabelValues().Dec()
 }
 
+func SetTriggerStatus(trigname, trignamespace string) {
+	triggerStatus.WithLabelValues(trigname, trignamespace).Inc()
+}
+
+func ResetTriggerStatus(trigname, trignamespace string) {
+	triggerStatus.WithLabelValues(trigname, trignamespace).Dec()
+}
+
+func IncreaseInprocessCount() {
+	mqtInprocessCount.WithLabelValues().Inc()
+}
+
+func DecreaseInprocessCount() {
+	mqtInprocessCount.WithLabelValues().Dec()
+}
+
 func IncreaseMessageCount(trigname, trignamespace string) {
 	messageCount.WithLabelValues(trigname, trignamespace).Inc()
 }
@@ -68,4 +98,6 @@ func init() {
 	registry.MustRegister(subscriptionCount)
 	registry.MustRegister(messageCount)
 	registry.MustRegister(messageLagCount)
+	registry.MustRegister(mqtInprocessCount)
+	registry.MustRegister(triggerStatus)
 }