prometheus for statistics

internal/rpc/msg/pull_message.go

@@ -8,6 +8,8 @@ import (
 	commonDB "Open_IM/pkg/common/db"
 	"Open_IM/pkg/common/log"
 	open_im_sdk "Open_IM/pkg/proto/sdk_ws"
+
+	promePkg "Open_IM/pkg/common/prometheus"
 )
 
 func (rpc *rpcChat) GetMaxAndMinSeq(_ context.Context, in *open_im_sdk.GetMaxAndMinSeqReq) (*open_im_sdk.GetMaxAndMinSeqResp, error) {
@@ -48,57 +50,62 @@ func (rpc *rpcChat) PullMessageBySeqList(_ context.Context, in *open_im_sdk.Pull
 	log.NewInfo(in.OperationID, "rpc PullMessageBySeqList is arriving", in.String())
 	resp := new(open_im_sdk.PullMessageBySeqListResp)
 	m := make(map[string]*open_im_sdk.MsgDataList)
-	//msgList, err := commonDB.DB.GetMsgBySeqList(in.UserID, in.SeqList, in.OperationID)
 	redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(in.UserID, in.SeqList, in.OperationID)
 	if err != nil {
 		if err != go_redis.Nil {
+			promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
 			log.Error(in.OperationID, "get message from redis exception", err.Error(), failedSeqList)
 		} else {
 			log.Debug(in.OperationID, "get message from redis is nil", failedSeqList)
 		}
 		msgList, err1 := commonDB.DB.GetMsgBySeqListMongo2(in.UserID, failedSeqList, in.OperationID)
 		if err1 != nil {
+			promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
 			log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err.Error())
 			resp.ErrCode = 201
 			resp.ErrMsg = err.Error()
 			return resp, nil
 		} else {
+			promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
 			redisMsgList = append(redisMsgList, msgList...)
 			resp.List = redisMsgList
 		}
 	} else {
+		promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
 		resp.List = redisMsgList
 	}
+
 	for k, v := range in.GroupSeqList {
 		x := new(open_im_sdk.MsgDataList)
 		redisMsgList, failedSeqList, err := commonDB.DB.GetMessageListBySeq(k, v.SeqList, in.OperationID)
 		if err != nil {
 			if err != go_redis.Nil {
+				promePkg.PromeAdd(promePkg.MsgPullFromRedisFailedCounter, len(failedSeqList))
 				log.Error(in.OperationID, "get message from redis exception", err.Error(), failedSeqList)
 			} else {
 				log.Debug(in.OperationID, "get message from redis is nil", failedSeqList)
 			}
 			msgList, err1 := commonDB.DB.GetSuperGroupMsgBySeqListMongo(k, failedSeqList, in.OperationID)
 			if err1 != nil {
+				promePkg.PromeAdd(promePkg.MsgPullFromMongoFailedCounter, len(failedSeqList))
 				log.Error(in.OperationID, "PullMessageBySeqList data error", in.String(), err.Error())
 				resp.ErrCode = 201
 				resp.ErrMsg = err.Error()
 				return resp, nil
 			} else {
+				promePkg.PromeAdd(promePkg.MsgPullFromMongoSuccessCounter, len(msgList))
 				redisMsgList = append(redisMsgList, msgList...)
 				x.MsgDataList = redisMsgList
 				m[k] = x
 			}
 		} else {
+			promePkg.PromeAdd(promePkg.MsgPullFromRedisSuccessCounter, len(redisMsgList))
 			x.MsgDataList = redisMsgList
 			m[k] = x
 		}
 	}
 	resp.GroupMsgDataList = m
-	//respSingleMsgFormat = singleMsgHandleByUser(SingleMsgFormat, in.UserID)
-	//respGroupMsgFormat = groupMsgHandleByUser(GroupMsgFormat)
 	return resp, nil
-
 }
 
 type MsgFormats []*open_im_sdk.MsgData

internal/rpc/msg/rpcChat.go

@@ -13,15 +13,14 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
+	promePkg "Open_IM/pkg/common/prometheus"
 	"google.golang.org/grpc"
 )
 
-var (
-	sendMsgSuccessCounter prometheus.Counter
-	sendMsgFailedCounter  prometheus.Counter
-)
+//var (
+//	sendMsgSuccessCounter prometheus.Counter
+//	sendMsgFailedCounter  prometheus.Counter
+//)
 
 type rpcChat struct {
 	rpcPort         int
@@ -55,14 +54,29 @@ func NewRpcChatServer(port int) *rpcChat {
 }
 
 func (rpc *rpcChat) initPrometheus() {
-	sendMsgSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "send_msg_success",
-		Help: "The number of send msg success",
-	})
-	sendMsgFailedCounter = promauto.NewCounter(prometheus.CounterOpts{
-		Name: "send_msg_failed",
-		Help: "The number of send msg failed",
-	})
+	//sendMsgSuccessCounter = promauto.NewCounter(prometheus.CounterOpts{
+	//	Name: "send_msg_success",
+	//	Help: "The number of send msg success",
+	//})
+	//sendMsgFailedCounter = promauto.NewCounter(prometheus.CounterOpts{
+	//	Name: "send_msg_failed",
+	//	Help: "The number of send msg failed",
+	//})
+	promePkg.NewMsgPullFromRedisSuccessCounter()
+	promePkg.NewMsgPullFromRedisFailedCounter()
+	promePkg.NewMsgPullFromMongoSuccessCounter()
+	promePkg.NewMsgPullFromMongoFailedCounter()
+
+	promePkg.NewSingleChatMsgRecvSuccessCounter()
+	promePkg.NewGroupChatMsgRecvSuccessCounter()
+	promePkg.NewWorkSuperGroupChatMsgRecvSuccessCounter()
+
+	promePkg.NewSingleChatMsgProcessSuccessCounter()
+	promePkg.NewSingleChatMsgProcessFailedCounter()
+	promePkg.NewGroupChatMsgProcessSuccessCounter()
+	promePkg.NewGroupChatMsgProcessFailedCounter()
+	promePkg.NewWorkSuperGroupChatMsgProcessSuccessCounter()
+	promePkg.NewWorkSuperGroupChatMsgProcessFailedCounter()
 }
 
 func (rpc *rpcChat) Run() {
@@ -97,9 +111,7 @@ func (rpc *rpcChat) Run() {
 		panic(utils.Wrap(err, "register chat module  rpc to etcd err"))
 	}
 	go rpc.runCh()
-	if config.Config.Prometheus.Enable {
-		rpc.initPrometheus()
-	}
+	rpc.initPrometheus()
 	err = srv.Serve(listener)
 	if err != nil {
 		log.Error("", "rpc rpcChat failed ", err.Error())

internal/rpc/msg/send_msg.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	promePkg "Open_IM/pkg/common/prometheus"
 	go_redis "github.com/go-redis/redis/v8"
 	"github.com/golang/protobuf/proto"
 )
@@ -270,6 +271,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 	}
 	switch pb.MsgData.SessionType {
 	case constant.SingleChatType:
+		promePkg.PromeInc(promePkg.SingleChatMsgRecvSuccessCounter)
 		// callback
 		t1 = time.Now()
 		callbackResp := callbackBeforeSendSingleMsg(pb)
@@ -282,6 +284,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 				callbackResp.ErrCode = 201
 			}
 			log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp)
+			promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
 		}
 		t1 = time.Now()
@@ -295,6 +298,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 			log.Info(pb.OperationID, "sendMsgToKafka ", " cost time: ", time.Since(t1))
 			if err1 != nil {
 				log.NewError(msgToMQSingle.OperationID, "kafka send msg err :RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String(), err1.Error())
+				promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
 				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 			}
 		}
@@ -304,6 +308,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 			log.Info(pb.OperationID, "sendMsgToKafka ", " cost time: ", time.Since(t1))
 			if err2 != nil {
 				log.NewError(msgToMQSingle.OperationID, "kafka send msg err:SendID", msgToMQSingle.MsgData.SendID, msgToMQSingle.String())
+				promePkg.PromeInc(promePkg.SingleChatMsgProcessFailedCounter)
 				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 			}
 		}
@@ -315,9 +320,11 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendSingleMsg resp: ", callbackResp)
 		}
 		log.Debug(pb.OperationID, "send msg cost time all: ", db.GetCurrentTimestampByMill()-newTime, pb.MsgData.ClientMsgID)
+		promePkg.PromeInc(promePkg.SingleChatMsgProcessSuccessCounter)
 		return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 	case constant.GroupChatType:
 		// callback
+		promePkg.PromeInc(promePkg.GroupChatMsgRecvSuccessCounter)
 		callbackResp := callbackBeforeSendGroupMsg(pb)
 		if callbackResp.ErrCode != 0 {
 			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendGroupMsg resp:", callbackResp)
@@ -327,10 +334,12 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 				callbackResp.ErrCode = 201
 			}
 			log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSingleMsg result", "end rpc and return", callbackResp)
+			promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
 		}
 		var memberUserIDList []string
 		if flag, errCode, errMsg, memberUserIDList = messageVerification(pb); !flag {
+			promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, errCode, errMsg, "", 0)
 		}
 		log.Debug(pb.OperationID, "GetGroupAllMember userID list", memberUserIDList, "len: ", len(memberUserIDList))
@@ -395,6 +404,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 		}
 		if !sendTag {
 			log.NewWarn(pb.OperationID, "send tag is ", sendTag)
+			promePkg.PromeInc(promePkg.GroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 		} else {
 			if pb.MsgData.ContentType == constant.AtText {
@@ -459,6 +469,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 				}()
 			}
 			log.Debug(pb.OperationID, "send msg cost time3 ", db.GetCurrentTimestampByMill()-newTime, pb.MsgData.ClientMsgID)
+			promePkg.PromeInc(promePkg.GroupChatMsgProcessSuccessCounter)
 			return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 		}
 	case constant.NotificationChatType:
@@ -481,6 +492,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 		log.Debug(pb.OperationID, "send msg cost time ", db.GetCurrentTimestampByMill()-newTime, pb.MsgData.ClientMsgID)
 		return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 	case constant.SuperGroupChatType:
+		promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgRecvSuccessCounter)
 		// callback
 		callbackResp := callbackBeforeSendGroupMsg(pb)
 		if callbackResp.ErrCode != 0 {
@@ -490,10 +502,12 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 			if callbackResp.ErrCode == 0 {
 				callbackResp.ErrCode = 201
 			}
+			promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
 			log.NewDebug(pb.OperationID, utils.GetSelfFuncName(), "callbackBeforeSendSuperGroupMsg result", "end rpc and return", callbackResp)
 			return returnMsg(&replay, pb, int32(callbackResp.ErrCode), callbackResp.ErrMsg, "", 0)
 		}
 		if flag, errCode, errMsg, _ = messageVerification(pb); !flag {
+			promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, errCode, errMsg, "", 0)
 		}
 		msgToMQSingle.MsgData = pb.MsgData
@@ -501,6 +515,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 		err1 := rpc.sendMsgToKafka(&msgToMQSingle, msgToMQSingle.MsgData.GroupID, constant.OnlineStatus)
 		if err1 != nil {
 			log.NewError(msgToMQSingle.OperationID, "kafka send msg err:RecvID", msgToMQSingle.MsgData.RecvID, msgToMQSingle.String())
+			promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessFailedCounter)
 			return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 		}
 		// callback
@@ -508,6 +523,7 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 		if callbackResp.ErrCode != 0 {
 			log.NewError(pb.OperationID, utils.GetSelfFuncName(), "callbackAfterSendSuperGroupMsg resp: ", callbackResp)
 		}
+		promePkg.PromeInc(promePkg.WorkSuperGroupChatMsgProcessSuccessCounter)
 		return returnMsg(&replay, pb, 0, "", msgToMQSingle.MsgData.ServerMsgID, msgToMQSingle.MsgData.SendTime)
 
 	default: