msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

prome

Browse Source
pull/351/head
wangchuxiao 2 years ago
parent a1f79f4508 6a6009827c
commit 024ee6e602
8 changed files with 178 additions and 153 deletions
Show Stats Download Patch File Download Diff File

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

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

1
internal/msg_transfer/logic/init.go
Unescape View File

@ -37,6 +37,7 @@ var (
func Init() {
cmdCh = make(chan Cmd2Value, 10000)
w = new(sync.Mutex)
initPrometheus()
persistentCH.Init() // ws2mschat save mysql
historyCH.Init(cmdCh) //
historyMongoCH.Init()

48
internal/msg_transfer/logic/persistent_msg_handler.go
Unescape View File

@ -18,13 +18,7 @@ import (
"github.com/Shopify/sarama"
"github.com/golang/protobuf/proto"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
msgInsertMysqlCounter prometheus.Counter
msgInsertFailedMysqlCounter prometheus.Counter
promePkg "Open_IM/pkg/common/prometheus"
)
type PersistentConsumerHandler struct {
@ -38,32 +32,18 @@ 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 {
pc.initPrometheus()
}
}
func (pc *PersistentConsumerHandler) initPrometheus() {
// counter
msgInsertMysqlCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "insert_mysql_msg_total",
Help: "The total number of msg insert mysql events",
})
msgInsertFailedMysqlCounter = promauto.NewCounter(prometheus.CounterOpts{
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 initPrometheus() {
promePkg.NewSeqGetSuccessCounter()
promePkg.NewSeqGetFailedCounter()
promePkg.NewSeqSetSuccessCounter()
promePkg.NewSeqSetFailedCounter()
promePkg.NewMsgInsertRedisSuccessCounter()
promePkg.NewMsgInsertRedisFailedCounter()
promePkg.NewMsgInsertMongoSuccessCounter()
promePkg.NewMsgInsertMongoFailedCounter()
}
func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMessage, msgKey string, _ sarama.ConsumerGroupSession) {
@ -97,14 +77,8 @@ func (pc *PersistentConsumerHandler) handleChatWs2Mysql(cMsg *sarama.ConsumerMes
log.NewInfo(msgFromMQ.OperationID, "msg_transfer msg persisting", string(msg))
if err = im_mysql_msg_model.InsertMessageToChatLog(msgFromMQ); err != nil {
log.NewError(msgFromMQ.OperationID, "Message insert failed", "err", err.Error(), "msg", msgFromMQ.String())
if config.Config.Prometheus.Enable {
msgInsertFailedMysqlCounter.Inc()
}
return
}
if config.Config.Prometheus.Enable {
msgInsertMysqlCounter.Inc()
}
}
}

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

@ -205,7 +205,7 @@ func (d *DataBases) GetMessageListBySeq(userID string, seqList []uint32, operati
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()
pipe := d.RDB.Pipeline()
var failedList []pbChat.MsgDataToMQ
@ -225,10 +225,10 @@ func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string,
}
}
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)
return err
return err, 0
}
func (d *DataBases) DeleteMessageFromCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) error {
ctx := context.Background()

209
pkg/common/db/batch_insert_chat.go
Unescape View File

@ -4,6 +4,7 @@ import (
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
promePkg "Open_IM/pkg/common/prometheus"
pbMsg "Open_IM/pkg/proto/msg"
"Open_IM/pkg/utils"
"context"
@ -14,10 +15,6 @@ import (
"go.mongodb.org/mongo-driver/mongo"
)
func (d *DataBases) BatchDeleteChat2DB(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) {
}
func (d *DataBases) BatchInsertChat2DB(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string, currentMaxSeq uint64) error {
newTime := getCurrentTimestampByMill()
if len(msgList) > GetSingleGocMsgNum() {
@ -85,10 +82,13 @@ func (d *DataBases) BatchInsertChat2DB(userID string, msgList []*pbMsg.MsgDataTo
sChat.Msg = msgListToMongo
log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo)
if _, err = c.InsertOne(ctx, &sChat); err != nil {
promePkg.PromeInc(promePkg.MsgInsertMongoFailedCounter)
log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "")
}
promePkg.PromeInc(promePkg.MsgInsertMongoSuccessCounter)
} else {
promePkg.PromeInc(promePkg.MsgInsertMongoFailedCounter)
log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter)
return utils.Wrap(err, "")
}
@ -101,9 +101,11 @@ func (d *DataBases) BatchInsertChat2DB(userID string, msgList []*pbMsg.MsgDataTo
sChat.Msg = msgListToMongoNext
log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext, "userID: ", userID)
if _, err = c.InsertOne(ctx, &sChat); err != nil {
promePkg.PromeInc(promePkg.MsgInsertMongoFailedCounter)
log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "")
}
promePkg.PromeInc(promePkg.MsgInsertMongoSuccessCounter)
}
log.Debug(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList))
return nil
@ -129,8 +131,11 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
log.Debug(operationID, "constant.SingleChatType lastMaxSeq before add ", currentMaxSeq, "userID ", insertID, err)
}
if err != nil && err != go_redis.Nil {
promePkg.PromeInc(promePkg.SeqGetFailedCounter)
return utils.Wrap(err, ""), 0
}
promePkg.PromeInc(promePkg.SeqGetSuccessCounter)
promePkg.SeqGetSuccessCounter.Inc()
lastMaxSeq := currentMaxSeq
for _, m := range msgList {
@ -142,9 +147,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, "SetMessageToCache ", insertID, len(msgList))
err = d.SetMessageToCache(msgList, insertID, operationID)
err, failedNum := d.SetMessageToCache(msgList, insertID, operationID)
if err != nil {
promePkg.PromeAdd(promePkg.MsgInsertRedisFailedCounter, failedNum)
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))
if msgList[0].MsgData.SessionType == constant.SuperGroupChatType {
@ -152,6 +160,11 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
} else {
err = d.SetUserMaxSeq(insertID, currentMaxSeq)
}
if err != nil {
promePkg.PromeInc(promePkg.SeqSetFailedCounter)
} else {
promePkg.PromeInc(promePkg.SeqSetSuccessCounter)
}
return utils.Wrap(err, ""), lastMaxSeq
}
@ -171,106 +184,100 @@ func (d *DataBases) BatchInsertChat2Cache(insertID string, msgList []*pbMsg.MsgD
// }
// return nil, lastMaxSeq
//}
func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error {
newTime := getCurrentTimestampByMill()
if len(msgList) > GetSingleGocMsgNum() {
return errors.New("too large")
}
isInit := false
currentMaxSeq, err := d.GetUserMaxSeq(userID)
if err == nil {
} else if err == go_redis.Nil {
isInit = true
currentMaxSeq = 0
} else {
return utils.Wrap(err, "")
}
var remain uint64
//
//func (d *DataBases) BatchInsertChat(userID string, msgList []*pbMsg.MsgDataToMQ, operationID string) error {
// newTime := getCurrentTimestampByMill()
// if len(msgList) > GetSingleGocMsgNum() {
// return errors.New("too large")
// }
// isInit := false
// currentMaxSeq, err := d.GetUserMaxSeq(userID)
// if err == nil {
//
// } else if err == go_redis.Nil {
// isInit = true
// currentMaxSeq = 0
// } else {
// return utils.Wrap(err, "")
// }
// var remain uint64
// //if currentMaxSeq < uint64(GetSingleGocMsgNum()) {
// // remain = uint64(GetSingleGocMsgNum()-1) - (currentMaxSeq % uint64(GetSingleGocMsgNum()))
// //} else {
// // remain = uint64(GetSingleGocMsgNum()) - ((currentMaxSeq - (uint64(GetSingleGocMsgNum()) - 1)) % uint64(GetSingleGocMsgNum()))
// //}
//
// blk0 := uint64(GetSingleGocMsgNum() - 1)
// if currentMaxSeq < uint64(GetSingleGocMsgNum()) {
// remain = uint64(GetSingleGocMsgNum()-1) - (currentMaxSeq % uint64(GetSingleGocMsgNum()))
// remain = blk0 - currentMaxSeq
// } else {
// remain = uint64(GetSingleGocMsgNum()) - ((currentMaxSeq - (uint64(GetSingleGocMsgNum()) - 1)) % uint64(GetSingleGocMsgNum()))
// excludeBlk0 := currentMaxSeq - blk0
// remain = (uint64(GetSingleGocMsgNum()) - (excludeBlk0 % uint64(GetSingleGocMsgNum()))) % uint64(GetSingleGocMsgNum())
// }
//
// insertCounter := uint64(0)
// msgListToMongo := make([]MsgInfo, 0)
// msgListToMongoNext := make([]MsgInfo, 0)
// seqUid := ""
// seqUidNext := ""
// log.Debug(operationID, "remain ", remain, "insertCounter ", insertCounter, "currentMaxSeq ", currentMaxSeq, userID, len(msgList))
// //4998 remain ==1
// //4999
// for _, m := range msgList {
// log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID)
// currentMaxSeq++
// sMsg := MsgInfo{}
// sMsg.SendTime = m.MsgData.SendTime
// m.MsgData.Seq = uint32(currentMaxSeq)
// if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
// return utils.Wrap(err, "")
// }
// if isInit {
// msgListToMongoNext = append(msgListToMongoNext, sMsg)
// seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
// log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
// continue
// }
// if insertCounter < remain {
// msgListToMongo = append(msgListToMongo, sMsg)
// insertCounter++
// seqUid = getSeqUid(userID, uint32(currentMaxSeq))
// log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
// } else {
// msgListToMongoNext = append(msgListToMongoNext, sMsg)
// seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
// 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.Background()
// c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
//
// if seqUid != "" {
// filter := bson.M{"uid": seqUid}
// log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo)
// err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err()
// if err != nil {
// log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter)
// return utils.Wrap(err, "")
// }
// }
// if seqUidNext != "" {
// filter := bson.M{"uid": seqUidNext}
// sChat := UserChat{}
// sChat.UID = seqUidNext
// sChat.Msg = msgListToMongoNext
// log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext)
// if _, err = c.InsertOne(ctx, &sChat); err != nil {
// log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
// return utils.Wrap(err, "")
// }
// }
// log.NewWarn(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList))
// return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), "")
//}
blk0 := uint64(GetSingleGocMsgNum() - 1)
if currentMaxSeq < uint64(GetSingleGocMsgNum()) {
remain = blk0 - currentMaxSeq
} else {
excludeBlk0 := currentMaxSeq - blk0
remain = (uint64(GetSingleGocMsgNum()) - (excludeBlk0 % uint64(GetSingleGocMsgNum()))) % uint64(GetSingleGocMsgNum())
}
insertCounter := uint64(0)
msgListToMongo := make([]MsgInfo, 0)
msgListToMongoNext := make([]MsgInfo, 0)
seqUid := ""
seqUidNext := ""
log.Debug(operationID, "remain ", remain, "insertCounter ", insertCounter, "currentMaxSeq ", currentMaxSeq, userID, len(msgList))
//4998 remain ==1
//4999
for _, m := range msgList {
log.Debug(operationID, "msg node ", m.String(), m.MsgData.ClientMsgID)
currentMaxSeq++
sMsg := MsgInfo{}
sMsg.SendTime = m.MsgData.SendTime
m.MsgData.Seq = uint32(currentMaxSeq)
if sMsg.Msg, err = proto.Marshal(m.MsgData); err != nil {
return utils.Wrap(err, "")
}
if isInit {
msgListToMongoNext = append(msgListToMongoNext, sMsg)
seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
log.Debug(operationID, "msgListToMongoNext ", seqUidNext, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
continue
}
if insertCounter < remain {
msgListToMongo = append(msgListToMongo, sMsg)
insertCounter++
seqUid = getSeqUid(userID, uint32(currentMaxSeq))
log.Debug(operationID, "msgListToMongo ", seqUid, m.MsgData.Seq, m.MsgData.ClientMsgID, insertCounter, remain)
} else {
msgListToMongoNext = append(msgListToMongoNext, sMsg)
seqUidNext = getSeqUid(userID, uint32(currentMaxSeq))
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.Background()
c := d.mongoClient.Database(config.Config.Mongo.DBDatabase).Collection(cChat)
if seqUid != "" {
filter := bson.M{"uid": seqUid}
log.NewDebug(operationID, "filter ", seqUid, "list ", msgListToMongo)
err := c.FindOneAndUpdate(ctx, filter, bson.M{"$push": bson.M{"msg": bson.M{"$each": msgListToMongo}}}).Err()
if err != nil {
log.Error(operationID, "FindOneAndUpdate failed ", err.Error(), filter)
return utils.Wrap(err, "")
}
}
if seqUidNext != "" {
filter := bson.M{"uid": seqUidNext}
sChat := UserChat{}
sChat.UID = seqUidNext
sChat.Msg = msgListToMongoNext
log.NewDebug(operationID, "filter ", seqUidNext, "list ", msgListToMongoNext)
if _, err = c.InsertOne(ctx, &sChat); err != nil {
log.NewError(operationID, "InsertOne failed", filter, err.Error(), sChat)
return utils.Wrap(err, "")
}
}
log.NewWarn(operationID, "batch mgo cost time ", getCurrentTimestampByMill()-newTime, userID, len(msgList))
return utils.Wrap(d.SetUserMaxSeq(userID, uint64(currentMaxSeq)), "")
}
//func (d *DataBases)setMessageToCache(msgList []*pbMsg.MsgDataToMQ, uid string) (err error) {
//
//}
func (d *DataBases) GetFromCacheAndInsertDB(msgUserIDPrefix string) {
//get value from redis
//batch insert to db
}

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

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

39
pkg/common/prometheus/gather.go
Unescape View File

@ -26,6 +26,11 @@ var (
GrpcRequestFailedCounter prometheus.Counter
SendMsgCounter prometheus.Counter
MsgInsertRedisSuccessCounter prometheus.Counter
MsgInsertRedisFailedCounter prometheus.Counter
MsgInsertMongoSuccessCounter prometheus.Counter
MsgInsertMongoFailedCounter prometheus.Counter
)
func NewUserLoginCounter() {
@ -34,6 +39,12 @@ func NewUserLoginCounter() {
Help: "The number of user login",
})
}
func NewUserRegisterCounter() {
UserRegisterCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "user_register",
Help: "The number of user register",
})
}
func NewUserRegisterCounter() {
UserRegisterCounter = promauto.NewCounter(prometheus.CounterOpts{
@ -116,3 +127,31 @@ func NewSendMsgCount() {
Help: "The number of send msg",
})
}
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",
})
}
func NewMsgInsertMongoSuccessCounter() {
MsgInsertMongoSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "msg_insert_mongo_success",
Help: "The number of successful insert msg to mongo",
})
}
func NewMsgInsertMongoFailedCounter() {
MsgInsertMongoFailedCounter = promauto.NewCounter(prometheus.CounterOpts{
Name: "msg_insert_mongo_failed",
Help: "The number of failed insert msg to mongo",
})
}

5
pkg/common/prometheus/prometheus.go
Unescape View File

@ -51,12 +51,17 @@ func PromeTheusMiddleware(c *gin.Context) {
func PromeInc(counter prometheus.Counter) {
if config.Config.Prometheus.Enable {
if counter != nil {
counter.Inc()
}
}
}
func PromeAdd(counter prometheus.Counter, add int) {
if config.Config.Prometheus.Enable {
if counter != nil {
counter.Add(float64(add))
}
}
}

Write Preview
Loading…
Cancel
Save