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

Log add automatic cutting function

Browse Source
pull/4/head
Gordon 4 years ago
parent 536f511b38
commit fbb079fb8c
6 changed files with 61 additions and 123 deletions
Show Stats Download Patch File Download Diff File

16
config/config.yaml
Unescape View File

@ -8,10 +8,10 @@ etcd:
etcdAddr: [ 127.0.0.1:2379 ]
mysql:
dbAddress: [ 127.0.0.1:3306 ]
dbUserName: root
dbPassword: 123456
dbDatabaseName: openIM
dbMysqlAddress: [ 127.0.0.1:3306 ]
dbMysqlUserName: root
dbMysqlPassword: 123456
dbMysqlDatabaseName: openIM
dbTableName: eMsg
dbMsgTableNum: 1
dbMaxOpenConns: 20
@ -90,6 +90,8 @@ rpcregistername:
log:
storageLocation: ../logs/
rotationTime: 12
remainRotationCount: 10
elasticSearchSwitch: false
elasticSearchAddr: [ 47.112.160.66:9201 ]
elasticSearchUser: ""
@ -127,4 +129,8 @@ tokenpolicy:
accessSecret: "open_im_server"
# Token effective time seconds as a unit
#Seven days 7*24*60*60
accessExpire: 604800
accessExpire: 604800
messagecallback:
callbackSwitch: false
callbackUrl: "http://www.xxx.com/msg/judge"

4
src/api/chat/send_msg.go
Unescape View File

@ -83,8 +83,8 @@ func UserSendMsg(c *gin.Context) {
log.Info("", "", "api UserSendMsg call end..., [data: %s] [reply: %s]", pbData.String(), reply.String())
c.JSON(http.StatusOK, gin.H{
"errCode": 0,
"errMsg": "",
"errCode": reply.ErrCode,
"errMsg": reply.ErrMsg,
"msgIncr": reply.MsgIncr,
"reqIdentifier": reply.ReqIdentifier,
"data": gin.H{

14
src/common/config/config.go
Unescape View File

@ -25,10 +25,10 @@ type config struct {
}
Mysql struct {
DBAddress []string `yaml:"dbAddress"`
DBUserName string `yaml:"dbUserName"`
DBPassword string `yaml:"dbPassword"`
DBDatabaseName string `yaml:"dbDatabaseName"`
DBAddress []string `yaml:"dbMysqlAddress"`
DBUserName string `yaml:"dbMysqlUserName"`
DBPassword string `yaml:"dbMysqlPassword"`
DBDatabaseName string `yaml:"dbMysqlDatabaseName"`
DBTableName string `yaml:"DBTableName"`
DBMsgTableNum int `yaml:"dbMsgTableNum"`
DBMaxOpenConns int `yaml:"dbMaxOpenConns"`
@ -77,6 +77,8 @@ type config struct {
}
Log struct {
StorageLocation string `yaml:"storageLocation"`
RotationTime int `yaml:"rotationTime"`
RemainRotationCount uint `yaml:"remainRotationCount"`
ElasticSearchSwitch bool `yaml:"elasticSearchSwitch"`
ElasticSearchAddr []string `yaml:"elasticSearchAddr"`
ElasticSearchUser string `yaml:"elasticSearchUser"`
@ -134,6 +136,10 @@ type config struct {
AccessSecret string `yaml:"accessSecret"`
AccessExpire int64 `yaml:"accessExpire"`
}
MessageCallBack struct {
CallbackSwitch bool `yaml:"callbackSwitch"`
CallbackUrl string `yaml:"callbackUrl"`
}
}
func init() {

61
src/common/log/logrus.go
Unescape View File

@ -44,56 +44,19 @@ func loggerInit(moduleName string) *Logger {
logger.AddHook(newEsHook(moduleName))
}
//Log file segmentation hook
hook := NewLfsHook(config.Config.Log.StorageLocation+time.Now().Format("2006-01-02")+".log", 0, 5, moduleName)
hook := NewLfsHook(time.Duration(config.Config.Log.RotationTime)*time.Hour, config.Config.Log.RemainRotationCount, moduleName)
logger.AddHook(hook)
return &Logger{
logger,
os.Getpid(),
}
}
func NewLfsHook(logName string, rotationTime time.Duration, maxRemainNum uint, moduleName string) logrus.Hook {
var fileNameSuffix string
if GetCurrentTimestamp() >= GetCurDayZeroTimestamp() && GetCurrentTimestamp() <= GetCurDayHalfTimestamp() {
fileNameSuffix = time.Now().Format("2006-01-02") + ".log"
} else {
fileNameSuffix = time.Now().Format("2006-01-02") + ".log"
}
writer, err := rotatelogs.New(
logName,
rotatelogs.WithRotationCount(maxRemainNum),
)
if err != nil {
panic(err)
}
writeInfo, err := rotatelogs.New(
config.Config.Log.StorageLocation+moduleName+"/info."+fileNameSuffix,
rotatelogs.WithRotationTime(time.Duration(60)*time.Second),
rotatelogs.WithRotationCount(maxRemainNum),
)
writeError, err := rotatelogs.New(
config.Config.Log.StorageLocation+moduleName+"/error."+fileNameSuffix,
rotatelogs.WithRotationTime(time.Minute),
rotatelogs.WithRotationCount(maxRemainNum),
)
writeDebug, err := rotatelogs.New(
config.Config.Log.StorageLocation+moduleName+"/debug."+fileNameSuffix,
rotatelogs.WithRotationCount(maxRemainNum),
)
writeWarn, err := rotatelogs.New(
config.Config.Log.StorageLocation+moduleName+"/warn."+fileNameSuffix,
rotatelogs.WithRotationTime(time.Minute),
rotatelogs.WithRotationCount(maxRemainNum),
)
if err != nil {
panic(err)
}
func NewLfsHook(rotationTime time.Duration, maxRemainNum uint, moduleName string) logrus.Hook {
lfsHook := lfshook.NewHook(lfshook.WriterMap{
logrus.DebugLevel: writeDebug,
logrus.InfoLevel: writeInfo,
logrus.WarnLevel: writeWarn,
logrus.ErrorLevel: writeError,
logrus.FatalLevel: writer,
logrus.PanicLevel: writer,
logrus.DebugLevel: initRotateLogs(rotationTime, maxRemainNum, "debug", moduleName),
logrus.InfoLevel: initRotateLogs(rotationTime, maxRemainNum, "info", moduleName),
logrus.WarnLevel: initRotateLogs(rotationTime, maxRemainNum, "warn", moduleName),
logrus.ErrorLevel: initRotateLogs(rotationTime, maxRemainNum, "error", moduleName),
}, &nested.Formatter{
TimestampFormat: "2006-01-02 15:04:05",
HideKeys: false,
@ -102,6 +65,18 @@ func NewLfsHook(logName string, rotationTime time.Duration, maxRemainNum uint, m
return lfsHook
}
func initRotateLogs(rotationTime time.Duration, maxRemainNum uint, level string, moduleName string) *rotatelogs.RotateLogs {
writer, err := rotatelogs.New(
config.Config.Log.StorageLocation+moduleName+"/"+level+"."+"%Y-%m-%d_%H-%M-%S",
rotatelogs.WithRotationTime(rotationTime),
rotatelogs.WithRotationCount(maxRemainNum),
)
if err != nil {
panic(err)
} else {
return writer
}
}
func Info(token, OperationID, format string, args ...interface{}) {
if token == "" && OperationID == "" {

41
src/msg_gateway/gate/rpc_server.go
Unescape View File

@ -77,18 +77,20 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
case constant.SyncSenderMsg:
log.InfoByKv("come sync", in.OperationID, "args", in.String())
RecvID = in.GetSendID()
for key, conn := range ws.wsUserToConn {
UIDAndPID := strings.Split(key, " ")
if UIDAndPID[0] == RecvID && utils.PlatformIDToName(in.GetPlatformID()) != UIDAndPID[1] {
resultCode := sendMsgToUser(conn, bMsg, in, UIDAndPID[1], UIDAndPID[0])
temp := &pbRelay.SingleMsgToUser{
ResultCode: resultCode,
RecvID: UIDAndPID[0],
RecvPlatFormID: utils.PlatformNameToID(UIDAndPID[1]),
if in.MsgFrom != constant.SysMsgType {
for key, conn := range ws.wsUserToConn {
UIDAndPID := strings.Split(key, " ")
if UIDAndPID[0] == RecvID && utils.PlatformIDToName(in.GetPlatformID()) != UIDAndPID[1] {
resultCode := sendMsgToUser(conn, bMsg, in, UIDAndPID[1], UIDAndPID[0])
temp := &pbRelay.SingleMsgToUser{
ResultCode: resultCode,
RecvID: UIDAndPID[0],
RecvPlatFormID: utils.PlatformNameToID(UIDAndPID[1]),
}
resp = append(resp, temp)
}
resp = append(resp, temp)
}
}
}
default:
log.InfoByKv("not come sync", in.OperationID, "args", in.String())
@ -120,25 +122,6 @@ func (r *RPCServer) MsgToUser(_ context.Context, in *pbRelay.MsgToUserReq) (*pbR
}, nil
}
//func (r *RPCServer) SendMsgByWS(_ context.Context, in *pbRelay.SendMsgByWSReq) (*pbRelay.MsgToUserResp, error) {
// log.InfoByKv("SendMsgByWS is arriving ", in.OperationID, "args", in.String())
// resp := new(pbRelay.MsgToUserResp)
// MsgId := ws.genMsgNum()
// pbData := pbMsg.WSToMsgSvrChatMsg{}
// pbData.SendID = in.SendID
// pbData.RecvID = in.RecvID
// pbData.MsgID = MsgId
// pbData.SessionType = in.SessionType
// pbData.MsgFrom = in.MsgFrom
// pbData.Content = in.Content
// pbData.ContentType = in.ContentType
// pbData.OperationID = in.OperationID
// pbData.SendTime = in.SendTime
// pbData.PlatformID = in.PlatformID
// pKafka.writeMsg(&pbData)
// return resp, nil
//}
func sendMsgToUser(conn *websocket.Conn, bMsg []byte, in *pbRelay.MsgToUserReq, RecvPlatForm, RecvID string) (ResultCode int64) {
err := ws.writeMsg(conn, websocket.TextMessage, bMsg)
if err != nil {

48
src/rpc/chat/chat/pull_message.go
Unescape View File

@ -1,4 +1,3 @@
//实现pb定义的rpc服务
package rpcChat
import (
@ -33,37 +32,6 @@ func (rpc *rpcChat) GetNewSeq(_ context.Context, in *pbMsg.GetNewSeqReq) (*pbMsg
}
//func (s *MsgServer) PullMessage(_ context.Context, in *pbMsg.PullMessageReq) (*pbMsg.PullMessageResp, error) {
// log.InfoByArgs(fmt.Sprintf("rpc pullMessage is arriving,args=%s", in.String()))
// resp := new(pbMsg.PullMessageResp)
// var respMsgFormat []*pbMsg.MsgFormat
// var respUserMsgFormat []*pbMsg.UserMsgFormat
// conn := db.NewDbConnection()
// rows, err := conn.Table("receive r").Select("c.sender_id,c.receiver_id,"+
// "c.msg_type,c.push_msg_type,c.chat_type,c.msg_id,c.send_content,r.seq,c.send_time,c.sender_nickname,c.receiver_nickname,c.sender_head_url,c.receiver_head_url").
// Joins("INNER JOIN chat_log c ON r.msg_id = c.msg_id AND r.user_id = ? AND seq BETWEEN ? AND ?",
// in.UserID, in.SeqBegin, in.SeqEnd).Rows()
// if err != nil {
// fmt.Printf("pullMsg data error: %v\n", err)
// resp.ErrCode = 1
// resp.ErrMsg = err.Error()
// return resp, nil
// }
// defer rows.Close()
// for rows.Next() {
// tempResp := new(pbMsg.MsgFormat)
// rows.Scan(&tempResp.SendID, &tempResp.RecvID, &tempResp.MsgType, &tempResp.PushMsgType, &tempResp.ChatType,
// &tempResp.MsgID, &tempResp.Msg, &tempResp.Seq, &tempResp.Time, &tempResp.SendNickName, &tempResp.RecvNickName,
// &tempResp.SendHeadUrl, &tempResp.RecvHeadUrl)
// respMsgFormat = append(respMsgFormat, tempResp)
// }
// respUserMsgFormat = msgHandleByUser(respMsgFormat, in.UserID)
// return &pbMsg.PullMessageResp{
// ErrCode: 0,
// ErrMsg: "",
// UserMsg: respUserMsgFormat,
// }, nil
//}
func (rpc *rpcChat) PullMessage(_ context.Context, in *pbMsg.PullMessageReq) (*pbMsg.PullMessageResp, error) {
log.InfoByKv("rpc pullMessage is arriving", in.OperationID, "args", in.String())
resp := new(pbMsg.PullMessageResp)
@ -91,7 +59,7 @@ func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.G
var userid string
var respMsgFormat []*pbMsg.GatherFormat
m := make(map[string]MsgFormats)
//将消息以用户为维度聚集
//Gather messages in the dimension of users
for _, v := range allMsg {
if v.RecvID != ownerId {
userid = v.RecvID
@ -105,7 +73,7 @@ func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.G
m[userid] = append(value, v)
}
}
//形成pb格式返回
//Return in pb format
for user, msg := range m {
tempUserMsg := new(pbMsg.GatherFormat)
tempUserMsg.ID = user
@ -118,9 +86,9 @@ func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.G
func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
var respMsgFormat []*pbMsg.GatherFormat
m := make(map[string]MsgFormats)
//将消息以用户为维度聚集
//Gather messages in the dimension of users
for _, v := range allMsg {
//获得群ID
//Get group ID
groupID := strings.Split(v.RecvID, " ")[1]
if value, ok := m[groupID]; !ok {
var t MsgFormats
@ -130,7 +98,7 @@ func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
}
}
//形成pb格式返回
//Return in pb format
for groupID, msg := range m {
tempUserMsg := new(pbMsg.GatherFormat)
tempUserMsg.ID = groupID
@ -143,17 +111,17 @@ func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
type MsgFormats []*pbMsg.MsgFormat
// 实现sort.Interface接口取元素数量方法
// Implement the sort.Interface interface to get the number of elements method
func (s MsgFormats) Len() int {
return len(s)
}
// 实现sort.Interface接口比较元素方法
//Implement the sort.Interface interface comparison element method
func (s MsgFormats) Less(i, j int) bool {
return s[i].SendTime < s[j].SendTime
}
// 实现sort.Interface接口交换元素方法
//Implement the sort.Interface interface exchange element method
func (s MsgFormats) Swap(i, j int) {
s[i], s[j] = s[j], s[i]
}

Write Preview
Loading…
Cancel
Save