blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-24 02:16:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

prometheus for statistics

Browse Source
This commit is contained in:
skiffer-git 2022-09-14 19:36:59 +08:00
parent 266a2b4c76
commit ebdb875c56
6 changed files with 145 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/cron_task/test/main.go
View File

@ -1,7 +1,6 @@
package main package main
import "Open_IM/pkg/common/db" //
//func main() {
func main() { // db.DB.BatchInsertChat()
db.DB.BatchInsertChat() //}
}

44
internal/msg_transfer/logic/persistent_msg_handler.go
View File

@ -18,13 +18,7 @@ import (
"github.com/Shopify/sarama" "github.com/Shopify/sarama"
"github.com/golang/protobuf/proto" "github.com/golang/protobuf/proto"
"github.com/prometheus/client_golang/prometheus" promePkg "Open_IM/pkg/common/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
msgInsertMysqlCounter prometheus.Counter
msgInsertFailedMysqlCounter prometheus.Counter
) )
type PersistentConsumerHandler struct { type PersistentConsumerHandler struct {
@ -38,32 +32,16 @@ func (pc *PersistentConsumerHandler) Init() {
pc.persistentConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0, pc.persistentConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{KafkaVersion: sarama.V2_0_0_0,
OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic}, OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false}, []string{config.Config.Kafka.Ws2mschat.Topic},
config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql) config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMySql)
if config.Config.Prometheus.Enable { pc.initPrometheus()
pc.initPrometheus()
}
} }
func (pc *PersistentConsumerHandler) initPrometheus() { func (pc *PersistentConsumerHandler) initPrometheus() {
// counter promePkg.NewSeqGetSuccessCounter()
msgInsertMysqlCounter = promauto.NewCounter(prometheus.CounterOpts{ promePkg.NewSeqGetFailedCounter()
Name: "insert_mysql_msg_total", promePkg.NewSeqSetSuccessCounter()
Help: "The total number of msg insert mysql events", promePkg.NewSeqSetFailedCounter()
}) promePkg.NewMsgInsertRedisSuccessCounter()
msgInsertFailedMysqlCounter = promauto.NewCounter(prometheus.CounterOpts{ promePkg.NewMsgInsertRedisFailedCounter()
Name: "insert_mysql_failed_msg_total",
Help: "The total number of msg insert mysql events",
})
// 启动计时器
// requestDurations := prometheus.NewHistogram(prometheus.HistogramOpts{
// Name: "http_request_duration_seconds",
// Help: "A histogram of the HTTP request durations in seconds.",
// Buckets: []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
// })
// 开始
// timer := prometheus.NewTimer(requestDurations)
// 停止
// timer.ObserveDuration()
} }
func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) { func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
@ -97,14 +75,8 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMes
log.NewInfo(msgFromMQ.OperationID, "msg_transfer msg persisting", string(msg)) log.NewInfo(msgFromMQ.OperationID, "msg_transfer msg persisting", string(msg))
if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil { if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil {
log.NewError(msgFromMQ.OperationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String()) log.NewError(msgFromMQ.OperationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String())
if config.Config.Prometheus.Enable {
msgInsertFailedMysqlCounter.Inc()
}
return return
} }
if config.Config.Prometheus.Enable {
msgInsertMysqlCounter.Inc()
}
} }
} }

6
pkg/common/db/RedisModel.go
View File

@ -205,7 +205,7 @@ func (d *DataBases) GetMessageListBySeq(userID string, seqList []uint32, operati
return seqMsg, failedSeqList, errResult return seqMsg, failedSeqList, errResult
} }
func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) error { func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) (error, int) {
ctx := context.Background() ctx := context.Background()
pipe := d.RDB.Pipeline() pipe := d.RDB.Pipeline()
var failedList []pbChat.MsgDataToMQ var failedList []pbChat.MsgDataToMQ
@ -225,10 +225,10 @@ func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string,
} }
} }
if len(failedList) != 0 { if len(failedList) != 0 {
return errors.New(fmt.Sprintf("set msg to cache failed, failed lists: %q,%s", failedList, operationID)) return errors.New(fmt.Sprintf("set msg to cache failed, failed lists: %q,%s", failedList, operationID)), len(failedList)
} }
_, err := pipe.Exec(ctx) _, err := pipe.Exec(ctx)
return err return err, 0
} }
func (d *DataBases) DeleteMessageFromCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) error { func (d *DataBases) DeleteMessageFromCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) error {
ctx := context.Background() ctx := context.Background()

200
pkg/common/db/batch_insert_chat.go
View File

@ -4,6 +4,7 @@ import (
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
promePkg "Open_IM/pkg/common/prometheus"
pbMsg "Open_IM/pkg/proto/msg" pbMsg "Open_IM/pkg/proto/msg"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
@ -129,8 +130,11 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
log.Debug(operationID, "constant.SingleChatType lastMaxSeq before add ", currentMaxSeq, "userID ", insertID, err) log.Debug(operationID, "constant.SingleChatType lastMaxSeq before add ", currentMaxSeq, "userID ", insertID, err)
} }
if err != nil && err != go_redis.Nil { if err != nil && err != go_redis.Nil {
promePkg.PromeInc(promePkg.SeqGetFailedCounter)
return utils.Wrap(err, ""), 0 return utils.Wrap(err, ""), 0
} }
promePkg.PromeInc(promePkg.SeqGetSuccessCounter)
promePkg.SeqGetSuccessCounter.Inc()
lastMaxSeq := currentMaxSeq lastMaxSeq := currentMaxSeq
for _, m := range msgList { for _, m := range msgList {
@ -142,9 +146,12 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
log.Debug(operationID, "cache msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", insertID, "seq: ", currentMaxSeq) log.Debug(operationID, "cache msg node ", m.String(), m.MsgData.ClientMsgID, "userID: ", insertID, "seq: ", currentMaxSeq)
} }
log.Debug(operationID, "SetMessageToCache ", insertID, len(msgList)) log.Debug(operationID, "SetMessageToCache ", insertID, len(msgList))
err = d.SetMessageToCache(msgList, insertID, operationID) err, failedNum := d.SetMessageToCache(msgList, insertID, operationID)
if err != nil { if err != nil {
promePkg.PromeAdd(promePkg.MsgInsertRedisFailedCounter, failedNum)
log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), insertID) log.Error(operationID, "setMessageToCache failed, continue ", err.Error(), len(msgList), insertID)
} else {
promePkg.PromeInc(promePkg.MsgInsertRedisSuccessCounter)
} }
log.Debug(operationID, "batch to redis cost time ", getCurrentTimestampByMill()-newTime, insertID, len(msgList)) log.Debug(operationID, "batch to redis cost time ", getCurrentTimestampByMill()-newTime, insertID, len(msgList))
if msgList[0].MsgData.SessionType == constant.SuperGroupChatType { if msgList[0].MsgData.SessionType == constant.SuperGroupChatType {
@ -152,6 +159,11 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
} else { } else {
err = d.SetUserMaxSeq(insertID, currentMaxSeq) err = d.SetUserMaxSeq(insertID, currentMaxSeq)
} }
if err != nil {
promePkg.PromeInc(promePkg.SeqSetFailedCounter)
} else {
promePkg.PromeInc(promePkg.SeqSetSuccessCounter)
}
return utils.Wrap(err, ""), lastMaxSeq return utils.Wrap(err, ""), lastMaxSeq
} }
@ -171,99 +183,99 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
// } // }
// return nil, lastMaxSeq // return nil, lastMaxSeq
//} //}
//
func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error { //func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error {
newTime := getCurrentTimestampByMill() // newTime := getCurrentTimestampByMill()
if len(msgList) > GetSingleGocMsgNum() { // if len(msgList) > GetSingleGocMsgNum() {
return errors.New("too large") // return errors.New("too large")
} // }
isInit := false // isInit := false
currentMaxSeq, err := d.GetUserMaxSeq(userID) // currentMaxSeq, err := d.GetUserMaxSeq(userID)
if err == nil { // if err == nil {
//
} else if err == go_redis.Nil { // } else if err == go_redis.Nil {
isInit = true // isInit = true
currentMaxSeq = 0 // currentMaxSeq = 0
} else { // } else {
return utils.Wrap(err, "") // return utils.Wrap(err, "")
} // }
var remain uint64 // var remain uint64
//if currentMaxSeq < uint64(GetSingleGocMsgNum()) { // //if currentMaxSeq < uint64(GetSingleGocMsgNum()) {
// remain = uint64(GetSingleGocMsgNum()-1) - (currentMaxSeq % uint64(GetSingleGocMsgNum())) // // remain = uint64(GetSingleGocMsgNum()-1) - (currentMaxSeq % uint64(GetSingleGocMsgNum()))
//} else { // //} else {
// remain = uint64(GetSingleGocMsgNum()) - ((currentMaxSeq - (uint64(GetSingleGocMsgNum()) - 1)) % uint64(GetSingleGocMsgNum())) // // remain = uint64(GetSingleGocMsgNum()) - ((currentMaxSeq - (uint64(GetSingleGocMsgNum()) - 1)) % uint64(GetSingleGocMsgNum()))
//} // //}
//
blk0 := uint64(GetSingleGocMsgNum() - 1) // blk0 := uint64(GetSingleGocMsgNum() - 1)
if currentMaxSeq < uint64(GetSingleGocMsgNum()) { // if currentMaxSeq < uint64(GetSingleGocMsgNum()) {
remain = blk0 - currentMaxSeq // remain = blk0 - currentMaxSeq
} else { // } else {
excludeBlk0 := currentMaxSeq - blk0 // excludeBlk0 := currentMaxSeq - blk0
remain = (uint64(GetSingleGocMsgNum()) - (excludeBlk0 % uint64(GetSingleGocMsgNum()))) % uint64(GetSingleGocMsgNum()) // remain = (uint64(GetSingleGocMsgNum()) - (excludeBlk0 % uint64(GetSingleGocMsgNum()))) % uint64(GetSingleGocMsgNum())
} // }
//
insertCounter := uint64(0) // insertCounter := uint64(0)
msgListToMongo := make([]MsgInfo, 0) // msgListToMongo := make([]MsgInfo, 0)
msgListToMongoNext := make([]MsgInfo, 0) // msgListToMongoNext := make([]MsgInfo, 0)
seqUid := "" // seqUid := ""
seqUidNext := "" // seqUidNext := ""
log.Debug(operationID, "remain ", remain, "insertCounter ", insertCounter, "currentMaxSeq ", currentMaxSeq, userID, len(msgList)) // log.Debug(operationID, "remain ", remain, "insertCounter ", insertCounter, "currentMaxSeq ", currentMaxSeq, userID, len(msgList))
//4998 remain ==1 // //4998 remain ==1
//4999 // //4999
for _, m := range msgList { // for _, m := range msgList {
log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID) // log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID)
currentMaxSeq++ // currentMaxSeq++
sMsg := MsgInfo{} // sMsg := MsgInfo{}
sMsg.SendTime = m.MsgData.SendTime // sMsg.SendTime = m.MsgData.SendTime
m.MsgData.Seq = uint32(currentMaxSeq) // m.MsgData.Seq = uint32(currentMaxSeq)
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil { // if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
return utils.Wrap(err, "") // return utils.Wrap(err, "")
} // }
if isInit { // if isInit {
msgListToMongoNext = append(msgListToMongoNext, sMsg) // msgListToMongoNext = append(msgListToMongoNext, sMsg)
seqUidNext = getSeqUid(userID, uint32(currentMaxSeq)) // seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) // log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
continue // continue
} // }
if insertCounter < remain { // if insertCounter < remain {
msgListToMongo = append(msgListToMongo, sMsg) // msgListToMongo = append(msgListToMongo, sMsg)
insertCounter++ // insertCounter++
seqUid = getSeqUid(userID, uint32(currentMaxSeq)) // seqUid = getSeqUid(userID, uint32(currentMaxSeq))
log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) // log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
} else { // } else {
msgListToMongoNext = append(msgListToMongoNext, sMsg) // msgListToMongoNext = append(msgListToMongoNext, sMsg)
seqUidNext = getSeqUid(userID, uint32(currentMaxSeq)) // seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain) // log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
} // }
} // }
// ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second) // // ctx, _ := context.WithTimeout(context.Background(), time.Duration(config.Config.Mongo.DBTimeout)*time.Second)
//
ctx := context.Background() // ctx := context.Background()
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat) // c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
//
if seqUid != "" { // if seqUid != "" {
filter := bson.M{"uid": seqUid} // filter := bson.M{"uid": seqUid}
log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo) // log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo)
err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err() // err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err()
if err != nil { // if err != nil {
log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter) // log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter)
return utils.Wrap(err, "") // return utils.Wrap(err, "")
} // }
} // }
if seqUidNext != "" { // if seqUidNext != "" {
filter := bson.M{"uid": seqUidNext} // filter := bson.M{"uid": seqUidNext}
sChat := UserChat{} // sChat := UserChat{}
sChat.UID = seqUidNext // sChat.UID = seqUidNext
sChat.Msg = msgListToMongoNext // sChat.Msg = msgListToMongoNext
log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext) // log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext)
if _, err = c.InsertOne(ctx, &sChat); err != nil { // if _, err = c.InsertOne(ctx, &sChat); err != nil {
log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat) // log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "") // return utils.Wrap(err, "")
} // }
} // }
log.NewWarn(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList)) // log.NewWarn(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList))
return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), "") // return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), "")
} //}
//func (d *DataBases)setMessageToCache(msgList []*pbMsg.MsgDataToMQ, uid string) (err error) { //func (d *DataBases)setMessageToCache(msgList []*pbMsg.MsgDataToMQ, uid string) (err error) {
// //

2
pkg/common/db/redisModel_test.go
View File

@ -71,7 +71,7 @@ func Test_NewSetMessageToCache(t *testing.T) {
data.AtUserIDList = []string{"1212", "23232"} data.AtUserIDList = []string{"1212", "23232"}
msg.MsgData = &data msg.MsgData = &data
messageList := []*pbChat.MsgDataToMQ{&msg} messageList := []*pbChat.MsgDataToMQ{&msg}
err := DB.SetMessageToCache(messageList, uid, "cacheTest") err, _ := DB.SetMessageToCache(messageList, uid, "cacheTest")
assert.Nil(t, err) assert.Nil(t, err)
} }

24
pkg/common/prometheus/gather.go
View File

@ -6,7 +6,6 @@ import (
) )
var ( var (
// user rpc
UserLoginCounter prometheus.Counter UserLoginCounter prometheus.Counter
UserRegisterCounter prometheus.Counter UserRegisterCounter prometheus.Counter
@ -14,6 +13,9 @@ var (
SeqGetFailedCounter prometheus.Counter SeqGetFailedCounter prometheus.Counter
SeqSetSuccessCounter prometheus.Counter SeqSetSuccessCounter prometheus.Counter
SeqSetFailedCounter prometheus.Counter SeqSetFailedCounter prometheus.Counter
MsgInsertRedisSuccessCounter prometheus.Counter
MsgInsertRedisFailedCounter prometheus.Counter
) )
func NewUserLoginCounter() { func NewUserLoginCounter() {
@ -22,6 +24,12 @@ func NewUserLoginCounter() {
Help: "The number of user login", Help: "The number of user login",
}) })
} }
func NewUserRegisterCounter() {
UserRegisterCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "user_register",
Help: "The number of user register",
})
}
func NewSeqGetSuccessCounter() { func NewSeqGetSuccessCounter() {
SeqGetSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{ SeqGetSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{
@ -48,3 +56,17 @@ func NewSeqSetFailedCounter() {
Help: "The number of failed set seq", Help: "The number of failed set seq",
}) })
} }
func NewMsgInsertRedisSuccessCounter() {
MsgInsertRedisSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "msg_insert_redis_success",
Help: "The number of successful insert msg to redis",
})
}
func NewMsgInsertRedisFailedCounter() {
MsgInsertRedisFailedCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "msg_insert_redis_failed",
Help: "The number of failed insert msg to redis",
})
}