feat: API RPC call statistics

cmd/openim-api/main.go

@@ -25,5 +25,4 @@ func main() {
 	if err := cmd.NewApiCmd().Exec(); err != nil {
 		program.ExitWithError(err)
 	}
-
 }

config/openim-crontask.yml

@@ -1,2 +1,3 @@
-chatRecordsClearTime: "0 2 * * *"
+cronExecuteTime: "0 2 * * *"
 retainChatRecords: 365
+fileExpireTime: 90

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/gorilla/websocket v1.5.1
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/mitchellh/mapstructure v1.5.0
-	github.com/openimsdk/protocol v0.0.69-alpha.17
+	github.com/openimsdk/protocol v0.0.69-alpha.24
 	github.com/openimsdk/tools v0.0.49-alpha.45
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0
@@ -175,5 +175,3 @@ require (
 	golang.org/x/crypto v0.21.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 )
-
-//replace github.com/openimsdk/protocol => /Users/chao/Desktop/project/protocol

go.sum

@@ -262,8 +262,8 @@ github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y=
 github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM=
 github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y=
 github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI=
-github.com/openimsdk/protocol v0.0.69-alpha.17 h1:pEag4ZdlovE+AyLsw1VYFU/3sk6ayvGdPzgufQfKf9M=
-github.com/openimsdk/protocol v0.0.69-alpha.17/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
+github.com/openimsdk/protocol v0.0.69-alpha.24 h1:TYcNJeWOTuE40UQ54eNPdDdy0KTOh9rAOgax8lCyhDc=
+github.com/openimsdk/protocol v0.0.69-alpha.24/go.mod h1:OZQA9FR55lseYoN2Ql1XAHYKHJGu7OMNkUbuekrKCM8=
 github.com/openimsdk/tools v0.0.49-alpha.45 h1:XIzCoef4myybOiIlGuRY9FTtGBisZFC4Uy4PhG0ZWQ0=
 github.com/openimsdk/tools v0.0.49-alpha.45/go.mod h1:HtSRjPTL8PsuZ+PhR5noqzrYBF0sdwW3/O/sWVucWg8=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=

internal/api/friend.go

@@ -57,7 +57,6 @@ func (o *FriendApi) GetFriendList(c *gin.Context) {
 
 func (o *FriendApi) GetDesignatedFriends(c *gin.Context) {
 	a2r.Call(relation.FriendClient.GetDesignatedFriends, o.Client, c)
-	//a2r.Call(relation.FriendClient.GetDesignatedFriends, o.Client, c, a2r.NewNilReplaceOption(relation.FriendClient.GetDesignatedFriends))
 }
 
 func (o *FriendApi) SetFriendRemark(c *gin.Context) {

internal/api/init.go

@@ -29,7 +29,6 @@ import (
 	"time"
 
 	kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister"
-	ginprom "github.com/openimsdk/open-im-server/v3/pkg/common/ginprometheus"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/tools/discovery"
 	"github.com/openimsdk/tools/errs"
@@ -72,10 +71,10 @@ func Start(ctx context.Context, index int, config *Config) error {
 				netDone <- struct{}{}
 				return
 			}
-			p := ginprom.NewPrometheus("app", prommetrics.GetGinCusMetrics("Api"))
-			p.SetListenAddress(fmt.Sprintf(":%d", prometheusPort))
-			if err = p.Use(router); err != nil && err != http.ErrServerClosed {
-				netErr = errs.WrapMsg(err, fmt.Sprintf("prometheus start err: %d", prometheusPort))
+			srv := http.NewServeMux()
+			srv.Handle(prommetrics.ApiPath, prommetrics.ApiHandler())
+			if err := http.ListenAndServe(fmt.Sprintf(":%d", prometheusPort), srv); err != nil && err != http.ErrServerClosed {
+				netErr = errs.WrapMsg(err, fmt.Sprintf("api prometheus start err: %d", prometheusPort))
 				netDone <- struct{}{}
 			}
 		}()

internal/api/router.go

@@ -2,12 +2,10 @@ package api
 
 import (
 	"fmt"
-	"net/http"
-	"strings"
-
 	"github.com/gin-gonic/gin"
 	"github.com/gin-gonic/gin/binding"
 	"github.com/go-playground/validator/v10"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
 	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 	"github.com/openimsdk/protocol/constant"
@@ -17,8 +15,25 @@ import (
 	"github.com/openimsdk/tools/mw"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"net/http"
+	"strings"
 )
 
+func prommetricsGin() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		c.Next()
+		path := c.FullPath()
+		if c.Writer.Status() == http.StatusNotFound {
+			prommetrics.HttpCall("<404>", c.Request.Method, c.Writer.Status())
+		} else {
+			prommetrics.HttpCall(path, c.Request.Method, c.Writer.Status())
+		}
+		if resp := apiresp.GetGinApiResponse(c); resp != nil {
+			prommetrics.APICall(path, c.Request.Method, resp.ErrCode)
+		}
+	}
+}
+
 func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.Engine {
 	disCov.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
 		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")))
@@ -37,7 +52,7 @@ func newGinRouter(disCov discovery.SvcDiscoveryRegistry, config *Config) *gin.En
 	authRpc := rpcclient.NewAuth(disCov, config.Share.RpcRegisterName.Auth)
 	thirdRpc := rpcclient.NewThird(disCov, config.Share.RpcRegisterName.Third, config.API.Prometheus.GrafanaURL)
 
-	r.Use(gin.Recovery(), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc))
+	r.Use(prommetricsGin(), gin.Recovery(), mw.CorsHandler(), mw.GinParseOperationID(), GinParseToken(authRpc))
 	u := NewUserApi(*userRpc)
 	m := NewMessageApi(messageRpc, userRpc, config.Share.IMAdminUserID)
 	userRouterGroup := r.Group("/user")

internal/msggateway/client.go

@@ -72,6 +72,8 @@ type Client struct {
 	closed         atomic.Bool
 	closedErr      error
 	token          string
+	//subLock        sync.Mutex
+	//subUserIDs     map[string]struct{}
 }
 
 // ResetClient updates the client's state with new connection and context information.

internal/msggateway/init.go

@@ -42,16 +42,15 @@ func Start(ctx context.Context, index int, conf *Config) error {
 	if err != nil {
 		return err
 	}
-	longServer, err := NewWsServer(
+	longServer := NewWsServer(
 		conf,
 		WithPort(wsPort),
 		WithMaxConnNum(int64(conf.MsgGateway.LongConnSvr.WebsocketMaxConnNum)),
 		WithHandshakeTimeout(time.Duration(conf.MsgGateway.LongConnSvr.WebsocketTimeout)*time.Second),
 		WithMessageMaxMsgLength(conf.MsgGateway.LongConnSvr.WebsocketMaxMsgLen),
 	)
-	if err != nil {
-		return err
-	}
+
+	go longServer.ChangeOnlineStatus(4)
 
 	hubServer := NewServer(rpcPort, longServer, conf)
 	netDone := make(chan error)

internal/msggateway/n_ws_server.go

@@ -54,13 +54,14 @@ type LongConnServer interface {
 }
 
 type WsServer struct {
-	msgGatewayConfig  *Config
-	port              int
-	wsMaxConnNum      int64
-	registerChan      chan *Client
-	unregisterChan    chan *Client
-	kickHandlerChan   chan *kickHandler
-	clients           *UserMap
+	msgGatewayConfig *Config
+	port             int
+	wsMaxConnNum     int64
+	registerChan     chan *Client
+	unregisterChan   chan *Client
+	kickHandlerChan  chan *kickHandler
+	clients          UserMap
+	//subscription      *Subscription
 	clientPool        sync.Pool
 	onlineUserNum     atomic.Int64
 	onlineUserConnNum atomic.Int64
@@ -90,18 +91,18 @@ func (ws *WsServer) SetDiscoveryRegistry(disCov discovery.SvcDiscoveryRegistry,
 	ws.disCov = disCov
 }
 
-func (ws *WsServer) SetUserOnlineStatus(ctx context.Context, client *Client, status int32) {
-	err := ws.userClient.SetUserStatus(ctx, client.UserID, status, client.PlatformID)
-	if err != nil {
-		log.ZWarn(ctx, "SetUserStatus err", err)
-	}
-	switch status {
-	case constant.Online:
-		ws.webhookAfterUserOnline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOnline, client.UserID, client.PlatformID, client.IsBackground, client.ctx.GetConnID())
-	case constant.Offline:
-		ws.webhookAfterUserOffline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOffline, client.UserID, client.PlatformID, client.ctx.GetConnID())
-	}
-}
+//func (ws *WsServer) SetUserOnlineStatus(ctx context.Context, client *Client, status int32) {
+//	err := ws.userClient.SetUserStatus(ctx, client.UserID, status, client.PlatformID)
+//	if err != nil {
+//		log.ZWarn(ctx, "SetUserStatus err", err)
+//	}
+//	switch status {
+//	case constant.Online:
+//		ws.webhookAfterUserOnline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOnline, client.UserID, client.PlatformID, client.IsBackground, client.ctx.GetConnID())
+//	case constant.Offline:
+//		ws.webhookAfterUserOffline(ctx, &ws.msgGatewayConfig.WebhooksConfig.AfterUserOffline, client.UserID, client.PlatformID, client.ctx.GetConnID())
+//	}
+//}
 
 func (ws *WsServer) UnRegister(c *Client) {
 	ws.unregisterChan <- c
@@ -119,7 +120,7 @@ func (ws *WsServer) GetUserPlatformCons(userID string, platform int) ([]*Client,
 	return ws.clients.Get(userID, platform)
 }
 
-func NewWsServer(msgGatewayConfig *Config, opts ...Option) (*WsServer, error) {
+func NewWsServer(msgGatewayConfig *Config, opts ...Option) *WsServer {
 	var config configs
 	for _, o := range opts {
 		o(&config)
@@ -141,10 +142,11 @@ func NewWsServer(msgGatewayConfig *Config, opts ...Option) (*WsServer, error) {
 		kickHandlerChan: make(chan *kickHandler, 1000),
 		validate:        v,
 		clients:         newUserMap(),
-		Compressor:      NewGzipCompressor(),
-		Encoder:         NewGobEncoder(),
-		webhookClient:   webhook.NewWebhookClient(msgGatewayConfig.WebhooksConfig.URL),
-	}, nil
+		//subscription:    newSubscription(),
+		Compressor:    NewGzipCompressor(),
+		Encoder:       NewGobEncoder(),
+		webhookClient: webhook.NewWebhookClient(msgGatewayConfig.WebhooksConfig.URL),
+	}
 }
 
 func (ws *WsServer) Run(done chan error) error {
@@ -278,11 +280,11 @@ func (ws *WsServer) registerClient(client *Client) {
 		}()
 	}
 
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		ws.SetUserOnlineStatus(client.ctx, client, constant.Online)
-	}()
+	//wg.Add(1)
+	//go func() {
+	//	defer wg.Done()
+	//	ws.SetUserOnlineStatus(client.ctx, client, constant.Online)
+	//}()
 
 	wg.Wait()
 
@@ -309,7 +311,7 @@ func getRemoteAdders(client []*Client) string {
 }
 
 func (ws *WsServer) KickUserConn(client *Client) error {
-	ws.clients.deleteClients(client.UserID, []*Client{client})
+	ws.clients.DeleteClients(client.UserID, []*Client{client})
 	return client.KickOnlineMessage()
 }
 
@@ -325,7 +327,7 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		if !clientOK {
 			return
 		}
-		ws.clients.deleteClients(newClient.UserID, oldClients)
+		ws.clients.DeleteClients(newClient.UserID, oldClients)
 		for _, c := range oldClients {
 			err := c.KickOnlineMessage()
 			if err != nil {
@@ -345,13 +347,13 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 
 func (ws *WsServer) unregisterClient(client *Client) {
 	defer ws.clientPool.Put(client)
-	isDeleteUser := ws.clients.delete(client.UserID, client.ctx.GetRemoteAddr())
+	isDeleteUser := ws.clients.DeleteClients(client.UserID, []*Client{client})
 	if isDeleteUser {
 		ws.onlineUserNum.Add(-1)
 		prommetrics.OnlineUserGauge.Dec()
 	}
 	ws.onlineUserConnNum.Add(-1)
-	ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
+	//ws.SetUserOnlineStatus(client.ctx, client, constant.Offline)
 	log.ZInfo(client.ctx, "user offline", "close reason", client.closedErr, "online user Num",
 		ws.onlineUserNum.Load(), "online user conn Num",
 		ws.onlineUserConnNum.Load(),

internal/msggateway/online.go

@@ -0,0 +1,112 @@
+package msggateway
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/binary"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/cachekey"
+	pbuser "github.com/openimsdk/protocol/user"
+	"github.com/openimsdk/tools/log"
+	"github.com/openimsdk/tools/mcontext"
+	"github.com/openimsdk/tools/utils/datautil"
+	"math/rand"
+	"strconv"
+	"time"
+)
+
+func (ws *WsServer) ChangeOnlineStatus(concurrent int) {
+	if concurrent < 1 {
+		concurrent = 1
+	}
+	const renewalTime = cachekey.OnlineExpire / 3
+	//const renewalTime = time.Second * 10
+	renewalTicker := time.NewTicker(renewalTime)
+
+	requestChs := make([]chan *pbuser.SetUserOnlineStatusReq, concurrent)
+	changeStatus := make([][]UserState, concurrent)
+
+	for i := 0; i < concurrent; i++ {
+		requestChs[i] = make(chan *pbuser.SetUserOnlineStatusReq, 64)
+		changeStatus[i] = make([]UserState, 0, 100)
+	}
+
+	mergeTicker := time.NewTicker(time.Second)
+
+	local2pb := func(u UserState) *pbuser.UserOnlineStatus {
+		return &pbuser.UserOnlineStatus{
+			UserID:  u.UserID,
+			Online:  u.Online,
+			Offline: u.Offline,
+		}
+	}
+
+	rNum := rand.Uint64()
+	pushUserState := func(us ...UserState) {
+		for _, u := range us {
+			sum := md5.Sum([]byte(u.UserID))
+			i := (binary.BigEndian.Uint64(sum[:]) + rNum) % uint64(concurrent)
+			changeStatus[i] = append(changeStatus[i], u)
+			status := changeStatus[i]
+			if len(status) == cap(status) {
+				req := &pbuser.SetUserOnlineStatusReq{
+					Status: datautil.Slice(status, local2pb),
+				}
+				changeStatus[i] = status[:0]
+				select {
+				case requestChs[i] <- req:
+				default:
+					log.ZError(context.Background(), "user online processing is too slow", nil)
+				}
+			}
+		}
+	}
+
+	pushAllUserState := func() {
+		for i, status := range changeStatus {
+			if len(status) == 0 {
+				continue
+			}
+			req := &pbuser.SetUserOnlineStatusReq{
+				Status: datautil.Slice(status, local2pb),
+			}
+			changeStatus[i] = status[:0]
+			select {
+			case requestChs[i] <- req:
+			default:
+				log.ZError(context.Background(), "user online processing is too slow", nil)
+			}
+		}
+	}
+
+	opIdCtx := mcontext.SetOperationID(context.Background(), "r"+strconv.FormatUint(rNum, 10))
+	doRequest := func(req *pbuser.SetUserOnlineStatusReq) {
+		ctx, cancel := context.WithTimeout(opIdCtx, time.Second*5)
+		defer cancel()
+		if _, err := ws.userClient.Client.SetUserOnlineStatus(ctx, req); err != nil {
+			log.ZError(ctx, "update user online status", err)
+		}
+	}
+
+	for i := 0; i < concurrent; i++ {
+		go func(ch <-chan *pbuser.SetUserOnlineStatusReq) {
+			for req := range ch {
+				doRequest(req)
+			}
+		}(requestChs[i])
+	}
+
+	for {
+		select {
+		case <-mergeTicker.C:
+			pushAllUserState()
+		case now := <-renewalTicker.C:
+			deadline := now.Add(-cachekey.OnlineExpire / 3)
+			users := ws.clients.GetAllUserStatus(deadline, now)
+			log.ZDebug(context.Background(), "renewal ticker", "deadline", deadline, "nowtime", now, "num", len(users))
+			pushUserState(users...)
+		case state := <-ws.clients.UserState():
+			log.ZDebug(context.Background(), "OnlineCache user online change", "userID", state.UserID, "online", state.Online, "offline", state.Offline)
+			pushUserState(state)
+		}
+	}
+}