docker-compose

internal/msg_transfer/logic/persistent_msg_handler.go

@@ -14,8 +14,16 @@ import (
 	"Open_IM/pkg/common/log"
 	pbMsg "Open_IM/pkg/proto/msg"
 	"Open_IM/pkg/utils"
+
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	msgInsertMysqlProcessed prometheus.Counter
 )
 
 type PersistentConsumerHandler struct {
@@ -29,7 +37,12 @@ func (pc *PersistentConsumerHandler) Init() {
 	pc.persistentConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
 		OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
 		config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql)
-
+	if config.Config.Prometheus.Enable {
+		msgInsertMysqlProcessed = promauto.NewCounter(prometheus.CounterOpts{
+			Name: "insert_mysql_msg_total",
+			Help: "The total number of msg insert mysql events",
+		})
+	}
 }
 
 func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
@@ -65,6 +78,9 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMes
 				log.NewError(msgFromMQ.OperationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String())
 				return
 			}
+			if config.Config.Prometheus.Enable {
+				msgInsertMysqlProcessed.Inc()
+			}
 		}
 
 	}