Repository: openimsdk/open-im-server
Pull request #1331 (2023-11-02 16:23:15)

refactor: crontask cmd
### summary

1. 重构了 cron_task 的启动代码, 加入了信号处理.
2. 通过分布式锁来实现 crontask 服务在分布式环境并发执行多次的问题。

> 通常为了 HA，cron 服务会多实例部署，但 cron 注册的任务同一时间只需要执行一次即可，无需多个实例执行。 

😁 (如果不需要这个分布式功能，我可以 reset 掉 )  

sweep-ai[bot]: ## Apply [Sweep Rules](https://docs.sweep.dev/usage/config#rules) to your PR?
- [ ] **Apply:**  Leftover TODOs in the code should be handled.
- [ ] **Apply:**  All new business logic should have corresponding unit tests in the tests/ directory.
- [ ] **Apply:**  Any clearly inefficient or repeated code should be optimized or refactored.
rfyiamcool: add unit test for cron with netlock.
cubxxw: @rfyiamcool  TODO
kubbot:  
> Bot detected the issue body's language is not English, translate it automatically. 👯👭🏻🧑‍🤝‍🧑👫🧑🏿‍🤝‍🧑🏻👩🏾‍🤝‍👨🏿👬🏿         
----    
@rfyiamcool EVERYTHING  
      

Patch:
diff --git a/internal/msggateway/n_ws_server.go b/internal/msggateway/n_ws_server.go
index 2466da2eb..a0cecd7f5 100644
--- a/internal/msggateway/n_ws_server.go
+++ b/internal/msggateway/n_ws_server.go
@@ -23,26 +23,22 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/OpenIMSDK/protocol/msggateway"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
-	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
-
-	"github.com/OpenIMSDK/protocol/constant"
-
-	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
-	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
-
+	"github.com/go-playground/validator/v10"
 	"github.com/redis/go-redis/v9"
+	"golang.org/x/sync/errgroup"
 
+	"github.com/OpenIMSDK/protocol/constant"
+	"github.com/OpenIMSDK/protocol/msggateway"
 	"github.com/OpenIMSDK/tools/discoveryregistry"
-
-	"github.com/go-playground/validator/v10"
-
 	"github.com/OpenIMSDK/tools/errs"
 	"github.com/OpenIMSDK/tools/log"
 	"github.com/OpenIMSDK/tools/utils"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/authverify"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
+	"github.com/openimsdk/open-im-server/v3/pkg/rpcclient"
 )
 
 type LongConnServer interface {
@@ -78,7 +74,6 @@ type WsServer struct {
 	onlineUserNum     atomic.Int64
 	onlineUserConnNum atomic.Int64
 	handshakeTimeout  time.Duration
-	hubServer         *Server
 	validate          *validator.Validate
 	cache             cache.MsgModel
 	userClient        *rpcclient.UserRpcClient
@@ -183,27 +178,39 @@ func (ws *WsServer) Run() error {
 	return http.ListenAndServe(":"+utils.IntToString(ws.port), nil) // Start listening
 }
 
+var concurrentRequest = 3
+
 func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *Client) error {
 	conns, err := ws.disCov.GetConns(ctx, config.Config.RpcRegisterName.OpenImMessageGatewayName)
 	if err != nil {
 		return err
 	}
+
+	wg := errgroup.Group{}
+	wg.SetLimit(concurrentRequest)
+
 	// Online push user online message to other node
 	for _, v := range conns {
+		v := v // safe closure var
 		if v.Target() == ws.disCov.GetSelfConnTarget() {
 			log.ZDebug(ctx, "Filter out this node", "node", v.Target())
 			continue
 		}
-		msgClient := msggateway.NewMsgGatewayClient(v)
-		_, err := msgClient.MultiTerminalLoginCheck(ctx, &msggateway.MultiTerminalLoginCheckReq{
-			UserID:     client.UserID,
-			PlatformID: int32(client.PlatformID), Token: client.token,
+
+		wg.Go(func() error {
+			msgClient := msggateway.NewMsgGatewayClient(v)
+			_, err := msgClient.MultiTerminalLoginCheck(ctx, &msggateway.MultiTerminalLoginCheckReq{
+				UserID:     client.UserID,
+				PlatformID: int32(client.PlatformID), Token: client.token,
+			})
+			if err != nil {
+				log.ZWarn(ctx, "MultiTerminalLoginCheck err", err, "node", v.Target())
+			}
+			return nil
 		})
-		if err != nil {
-			log.ZWarn(ctx, "MultiTerminalLoginCheck err", err, "node", v.Target())
-			continue
-		}
 	}
+
+	_ = wg.Wait()
 	return nil
 }
 
@@ -289,70 +296,72 @@ func (ws *WsServer) multiTerminalLoginChecker(clientOK bool, oldClients []*Clien
 		}
 		fallthrough
 	case constant.AllLoginButSameTermKick:
-		if clientOK {
-			isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients)
-			if isDeleteUser {
-				ws.onlineUserNum.Add(-1)
-			}
-			for _, c := range oldClients {
-				err := c.KickOnlineMessage()
-				if err != nil {
-					log.ZWarn(c.ctx, "KickOnlineMessage", err)
-				}
+		if !clientOK {
+			return
+		}
+
+		isDeleteUser := ws.clients.deleteClients(newClient.UserID, oldClients)
+		if isDeleteUser {
+			ws.onlineUserNum.Add(-1)
+		}
+		for _, c := range oldClients {
+			err := c.KickOnlineMessage()
+			if err != nil {
+				log.ZWarn(c.ctx, "KickOnlineMessage", err)
 			}
-			m, err := ws.cache.GetTokensWithoutError(
+		}
+		m, err := ws.cache.GetTokensWithoutError(
+			newClient.ctx,
+			newClient.UserID,
+			newClient.PlatformID,
+		)
+		if err != nil && err != redis.Nil {
+			log.ZWarn(
 				newClient.ctx,
+				"get token from redis err",
+				err,
+				"userID",
 				newClient.UserID,
+				"platformID",
 				newClient.PlatformID,
 			)
-			if err != nil && err != redis.Nil {
-				log.ZWarn(
-					newClient.ctx,
-					"get token from redis err",
-					err,
-					"userID",
-					newClient.UserID,
-					"platformID",
-					newClient.PlatformID,
-				)
-				return
-			}
-			if m == nil {
-				log.ZWarn(
-					newClient.ctx,
-					"m is nil",
-					errors.New("m is nil"),
-					"userID",
-					newClient.UserID,
-					"platformID",
-					newClient.PlatformID,
-				)
-				return
-			}
-			log.ZDebug(
+			return
+		}
+		if m == nil {
+			log.ZWarn(
 				newClient.ctx,
-				"get token from redis",
+				"m is nil",
+				errors.New("m is nil"),
 				"userID",
 				newClient.UserID,
 				"platformID",
 				newClient.PlatformID,
-				"tokenMap",
-				m,
 			)
-
-			for k := range m {
-				if k != newClient.ctx.GetToken() {
-					m[k] = constant.KickedToken
-				}
-			}
-			log.ZDebug(newClient.ctx, "set token map is ", "token map", m, "userID",
-				newClient.UserID, "token", newClient.ctx.GetToken())
-			err = ws.cache.SetTokenMapByUidPid(newClient.ctx, newClient.UserID, newClient.PlatformID, m)
-			if err != nil {
-				log.ZWarn(newClient.ctx, "SetTokenMapByUidPid err", err, "userID", newClient.UserID, "platformID", newClient.PlatformID)
-				return
+			return
+		}
+		log.ZDebug(
+			newClient.ctx,
+			"get token from redis",
+			"userID",
+			newClient.UserID,
+			"platformID",
+			newClient.PlatformID,
+			"tokenMap",
+			m,
+		)
+
+		for k := range m {
+			if k != newClient.ctx.GetToken() {
+				m[k] = constant.KickedToken
 			}
 		}
+		log.ZDebug(newClient.ctx, "set token map is ", "token map", m, "userID",
+			newClient.UserID, "token", newClient.ctx.GetToken())
+		err = ws.cache.SetTokenMapByUidPid(newClient.ctx, newClient.UserID, newClient.PlatformID, m)
+		if err != nil {
+			log.ZWarn(newClient.ctx, "SetTokenMapByUidPid err", err, "userID", newClient.UserID, "platformID", newClient.PlatformID)
+			return
+		}
 	}
 }
 
@@ -404,7 +413,7 @@ func (ws *WsServer) wsHandler(w http.ResponseWriter, r *http.Request) {
 		httpError(connContext, errs.ErrConnArgsErr)
 		return
 	}
-	if err := authverify.WsVerifyToken(token, userID, platformID); err != nil {
+	if err = authverify.WsVerifyToken(token, userID, platformID); err != nil {
 		httpError(connContext, err)
 		return
 	}
diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go
index bebf6819a..4ce015543 100644
--- a/internal/msgtransfer/init.go
+++ b/internal/msgtransfer/init.go
@@ -21,14 +21,13 @@ import (
 	"net/http"
 	"sync"
 
+	"github.com/OpenIMSDK/tools/mw"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
-	"github.com/OpenIMSDK/tools/mw"
-
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 	"github.com/openimsdk/open-im-server/v3/pkg/common/db/controller"
diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go
index b4556634c..b019b0120 100644
--- a/internal/msgtransfer/online_history_msg_handler.go
+++ b/internal/msgtransfer/online_history_msg_handler.go
@@ -427,49 +427,62 @@ func (och *OnlineHistoryRedisConsumerHandler) ConsumeClaim(
 			break
 		}
 	}
-	rwLock := new(sync.RWMutex)
 	log.ZDebug(context.Background(), "online new session msg come", "highWaterMarkOffset",
 		claim.HighWaterMarkOffset(), "topic", claim.Topic(), "partition", claim.Partition())
-	cMsg := make([]*sarama.ConsumerMessage, 0, 1000)
-	t := time.NewTicker(time.Millisecond * 100)
+
+	split := 1000
+	rwLock := new(sync.RWMutex)
+	messages := make([]*sarama.ConsumerMessage, 0, 1000)
+	ticker := time.NewTicker(time.Millisecond * 100)
+
 	go func() {
 		for {
 			select {
-			case <-t.C:
-				if len(cMsg) > 0 {
-					rwLock.Lock()
-					ccMsg := make([]*sarama.ConsumerMessage, 0, 1000)
-					for _, v := range cMsg {
-						ccMsg = append(ccMsg, v)
-					}
-					cMsg = make([]*sarama.ConsumerMessage, 0, 1000)
-					rwLock.Unlock()
-					split := 1000
-					ctx := mcontext.WithTriggerIDContext(context.Background(), utils.OperationIDGenerator())
-					log.ZDebug(ctx, "timer trigger msg consumer start", "length", len(ccMsg))
-					for i := 0; i < len(ccMsg)/split; i++ {
-						// log.Debug()
-						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-							ctx: ctx, cMsgList: ccMsg[i*split : (i+1)*split],
-						}}
-					}
-					if (len(ccMsg) % split) > 0 {
-						och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
-							ctx: ctx, cMsgList: ccMsg[split*(len(ccMsg)/split):],
-						}}
-					}
-					log.ZDebug(ctx, "timer trigger msg consumer end", "length", len(ccMsg))
+			case <-ticker.C:
+				if len(messages) == 0 {
+					continue
+				}
+
+				rwLock.Lock()
+				buffer := make([]*sarama.ConsumerMessage, 0, len(messages))
+				buffer = append(buffer, messages...)
+
+				// reuse slice, set cap to 0
+				messages = messages[:0]
+				rwLock.Unlock()
+
+				start := time.Now()
+				ctx := mcontext.WithTriggerIDContext(context.Background(), utils.OperationIDGenerator())
+				log.ZDebug(ctx, "timer trigger msg consumer start", "length", len(buffer))
+				for i := 0; i < len(buffer)/split; i++ {
+					och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
+						ctx: ctx, cMsgList: buffer[i*split : (i+1)*split],
+					}}
+				}
+				if (len(buffer) % split) > 0 {
+					och.msgDistributionCh <- Cmd2Value{Cmd: ConsumerMsgs, Value: TriggerChannelValue{
+						ctx: ctx, cMsgList: buffer[split*(len(buffer)/split):],
+					}}
 				}
+
+				log.ZDebug(ctx, "timer trigger msg consumer end",
+					"length", len(buffer), "time_cost", time.Since(start),
+				)
 			}
 		}
 	}()
+
 	for msg := range claim.Messages() {
-		rwLock.Lock()
-		if len(msg.Value) != 0 {
-			cMsg = append(cMsg, msg)
+		if len(msg.Value) == 0 {
+			continue
 		}
+
+		rwLock.Lock()
+		messages = append(messages, msg)
 		rwLock.Unlock()
+
 		sess.MarkMessage(msg, "")
 	}
+
 	return nil
 }
diff --git a/internal/tools/cron_task.go b/internal/tools/cron_task.go
index 6f4803628..e22504bbb 100644
--- a/internal/tools/cron_task.go
+++ b/internal/tools/cron_task.go
@@ -17,13 +17,18 @@ package tools
 import (
 	"context"
 	"fmt"
-	"sync"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
 
+	"github.com/redis/go-redis/v9"
 	"github.com/robfig/cron/v3"
 
 	"github.com/OpenIMSDK/tools/log"
 
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
 )
 
 func StartTask() error {
@@ -32,23 +37,75 @@ func StartTask() error {
 	if err != nil {
 		return err
 	}
-	msgTool.ConvertTools()
-	c := cron.New()
-	var wg sync.WaitGroup
-	wg.Add(1)
+
+	msgTool.convertTools()
+
+	rdb, err := cache.NewRedis()
+	if err != nil {
+		return err
+	}
+
+	// register cron tasks
+	var crontab = cron.New()
 	log.ZInfo(context.Background(), "start chatRecordsClearTime cron task", "cron config", config.Config.ChatRecordsClearTime)
-	_, err = c.AddFunc(config.Config.ChatRecordsClearTime, msgTool.AllConversationClearMsgAndFixSeq)
+	_, err = crontab.AddFunc(config.Config.ChatRecordsClearTime, cronWrapFunc(rdb, "cron_clear_msg_and_fix_seq", msgTool.AllConversationClearMsgAndFixSeq))
 	if err != nil {
 		log.ZError(context.Background(), "start allConversationClearMsgAndFixSeq cron failed", err)
 		panic(err)
 	}
+
 	log.ZInfo(context.Background(), "start msgDestruct cron task", "cron config", config.Config.MsgDestructTime)
-	_, err = c.AddFunc(config.Config.MsgDestructTime, msgTool.ConversationsDestructMsgs)
+	_, err = crontab.AddFunc(config.Config.MsgDestructTime, cronWrapFunc(rdb, "cron_conversations_destruct_msgs", msgTool.ConversationsDestructMsgs))
 	if err != nil {
 		log.ZError(context.Background(), "start conversationsDestructMsgs cron failed", err)
 		panic(err)
 	}
-	c.Start()
-	wg.Wait()
+
+	// start crontab
+	crontab.Start()
+
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	<-sigs
+
+	// stop crontab, Wait for the running task to exit.
+	ctx := crontab.Stop()
+
+	select {
+	case <-ctx.Done():
+		// graceful exit
+
+	case <-time.After(15 * time.Second):
+		// forced exit on timeout
+	}
+
 	return nil
 }
+
+// netlock redis lock.
+func netlock(rdb redis.UniversalClient, key string, ttl time.Duration) bool {
+	value := "used"
+	ok, err := rdb.SetNX(context.Background(), key, value, ttl).Result() // nolint
+	if err != nil {
+		// when err is about redis server, return true.
+		return false
+	}
+
+	return ok
+}
+
+func cronWrapFunc(rdb redis.UniversalClient, key string, fn func()) func() {
+	enableCronLocker := config.Config.EnableCronLocker
+	return func() {
+		// if don't enable cron-locker, call fn directly.
+		if !enableCronLocker {
+			fn()
+			return
+		}
+
+		// when acquire redis lock, call fn().
+		if netlock(rdb, key, 5*time.Second) {
+			fn()
+		}
+	}
+}
diff --git a/internal/tools/msg_doc_convert.go b/internal/tools/msg_doc_convert.go
index aa24d385f..758625be1 100644
--- a/internal/tools/msg_doc_convert.go
+++ b/internal/tools/msg_doc_convert.go
@@ -22,7 +22,7 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/msgprocessor"
 )
 
-func (c *MsgTool) ConvertTools() {
+func (c *MsgTool) convertTools() {
 	ctx := mcontext.NewCtx("convert")
 	conversationIDs, err := c.conversationDatabase.GetAllConversationIDs(ctx)
 	if err != nil {
diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go
index 5309f9913..94688b0fb 100644
--- a/pkg/common/config/config.go
+++ b/pkg/common/config/config.go
@@ -232,6 +232,7 @@ type configStruct struct {
 	ChatRecordsClearTime              string `yaml:"chatRecordsClearTime"`
 	MsgDestructTime                   string `yaml:"msgDestructTime"`
 	Secret                            string `yaml:"secret"`
+	EnableCronLocker                  bool   `yaml:"enableCronLocker"`
 	TokenPolicy                       struct {
 		Expire int64 `yaml:"expire"`
 	} `yaml:"tokenPolicy"`
diff --git a/pkg/common/db/cache/init_redis.go b/pkg/common/db/cache/init_redis.go
index 1a5507f89..77b38d9b7 100644
--- a/pkg/common/db/cache/init_redis.go
+++ b/pkg/common/db/cache/init_redis.go
@@ -28,12 +28,21 @@ import (
 	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
 )
 
+var (
+	// singleton pattern.
+	redisClient redis.UniversalClient
+)
+
 const (
 	maxRetry = 10 // number of retries
 )
 
 // NewRedis Initialize redis connection.
 func NewRedis() (redis.UniversalClient, error) {
+	if redisClient != nil {
+		return redisClient, nil
+	}
+
 	if len(config.Config.Redis.Address) == 0 {
 		return nil, errors.New("redis address is empty")
 	}
@@ -66,5 +75,6 @@ func NewRedis() (redis.UniversalClient, error) {
 		return nil, fmt.Errorf("redis ping %w", err)
 	}
 
+	redisClient = rdb
 	return rdb, err
 }
diff --git a/pkg/common/ginprometheus/ginprometheus.go b/pkg/common/ginprometheus/ginprometheus.go
index 50a6c3a2c..a325595d6 100644
--- a/pkg/common/ginprometheus/ginprometheus.go
+++ b/pkg/common/ginprometheus/ginprometheus.go
@@ -3,7 +3,7 @@ package ginprometheus
 import (
 	"bytes"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"os"
 	"strconv"
@@ -17,40 +17,42 @@ import (
 var defaultMetricPath = "/metrics"
 
 // counter, counter_vec, gauge, gauge_vec,
-// histogram, histogram_vec, summary, summary_vec
-var reqCnt = &Metric{
-	ID:          "reqCnt",
-	Name:        "requests_total",
-	Description: "How many HTTP requests processed, partitioned by status code and HTTP method.",
-	Type:        "counter_vec",
-	Args:        []string{"code", "method", "handler", "host", "url"}}
-
-var reqDur = &Metric{
-	ID:          "reqDur",
-	Name:        "request_duration_seconds",
-	Description: "The HTTP request latencies in seconds.",
-	Type:        "histogram_vec",
-	Args:        []string{"code", "method", "url"},
-}
+// histogram, histogram_vec, summary, summary_vec.
+var (
+	reqCounter = &Metric{
+		ID:          "reqCnt",
+		Name:        "requests_total",
+		Description: "How many HTTP requests processed, partitioned by status code and HTTP method.",
+		Type:        "counter_vec",
+		Args:        []string{"code", "method", "handler", "host", "url"}}
+
+	reqDuration = &Metric{
+		ID:          "reqDur",
+		Name:        "request_duration_seconds",
+		Description: "The HTTP request latencies in seconds.",
+		Type:        "histogram_vec",
+		Args:        []string{"code", "method", "url"},
+	}
 
-var resSz = &Metric{
-	ID:          "resSz",
-	Name:        "response_size_bytes",
-	Description: "The HTTP response sizes in bytes.",
-	Type:        "summary"}
-
-var reqSz = &Metric{
-	ID:          "reqSz",
-	Name:        "request_size_bytes",
-	Description: "The HTTP request sizes in bytes.",
-	Type:        "summary"}
-
-var standardMetrics = []*Metric{
-	reqCnt,
-	reqDur,
-	resSz,
-	reqSz,
-}
+	resSize = &Metric{
+		ID:          "resSz",
+		Name:        "response_size_bytes",
+		Description: "The HTTP response sizes in bytes.",
+		Type:        "summary"}
+
+	reqSize = &Metric{
+		ID:          "reqSz",
+		Name:        "request_size_bytes",
+		Description: "The HTTP request sizes in bytes.",
+		Type:        "summary"}
+
+	standardMetrics = []*Metric{
+		reqCounter,
+		reqDuration,
+		resSize,
+		reqSize,
+	}
+)
 
 /*
 RequestCounterURLLabelMappingFn is a function which can be supplied to the middleware to control
@@ -74,7 +76,7 @@ which would map "/customer/alice" and "/customer/bob" to their template "/custom
 type RequestCounterURLLabelMappingFn func(c *gin.Context) string
 
 // Metric is a definition for the name, description, type, ID, and
-// prometheus.Collector type (i.e. CounterVec, Summary, etc) of each metric
+// prometheus.Collector type (i.e. CounterVec, Summary, etc) of each metric.
 type Metric struct {
 	MetricCollector prometheus.Collector
 	ID              string
@@ -84,7 +86,7 @@ type Metric struct {
 	Args            []string
 }
 
-// Prometheus contains the metrics gathered by the instance and its path
+// Prometheus contains the metrics gathered by the instance and its path.
 type Prometheus struct {
 	reqCnt        *prometheus.CounterVec
 	reqDur        *prometheus.HistogramVec
@@ -102,7 +104,7 @@ type Prometheus struct {
 	URLLabelFromContext string
 }
 
-// PrometheusPushGateway contains the configuration for pushing to a Prometheus pushgateway (optional)
+// PrometheusPushGateway contains the configuration for pushing to a Prometheus pushgateway (optional).
 type PrometheusPushGateway struct {
 
 	// Push interval in seconds
@@ -112,7 +114,7 @@ type PrometheusPushGateway struct {
 	// where JOBNAME can be any string of your choice
 	PushGatewayURL string
 
-	// Local metrics URL where metrics are fetched from, this could be ommited in the future
+	// Local metrics URL where metrics are fetched from, this could be omitted in the future
 	// if implemented using prometheus common/expfmt instead
 	MetricsURL string
 
@@ -120,9 +122,11 @@ type PrometheusPushGateway struct {
 	Job string
 }
 
-// NewPrometheus generates a new set of metrics with a certain subsystem name
+// NewPrometheus generates a new set of metrics with a certain subsystem name.
 func NewPrometheus(subsystem string, customMetricsList ...[]*Metric) *Prometheus {
-	subsystem = "app"
+	if subsystem == "" {
+		subsystem = "app"
+	}
 
 	var metricsList []*Metric
 
@@ -131,16 +135,13 @@ func NewPrometheus(subsystem string, customMetricsList ...[]*Metric) *Prometheus
 	} else if len(customMetricsList) == 1 {
 		metricsList = customMetricsList[0]
 	}
-
-	for _, metric := range standardMetrics {
-		metricsList = append(metricsList, metric)
-	}
+	metricsList = append(metricsList, standardMetrics...)
 
 	p := &Prometheus{
 		MetricsList: metricsList,
 		MetricsPath: defaultMetricPath,
 		ReqCntURLLabelMappingFn: func(c *gin.Context) string {
-			return c.Request.URL.Path
+			return c.FullPath() // e.g. /user/:id , /user/:id/info
 		},
 	}
 
@@ -150,7 +151,7 @@ func NewPrometheus(subsystem string, customMetricsList ...[]*Metric) *Prometheus
 }
 
 // SetPushGateway sends metrics to a remote pushgateway exposed on pushGatewayURL
-// every pushIntervalSeconds. Metrics are fetched from metricsURL
+// every pushIntervalSeconds. Metrics are fetched from metricsURL.
 func (p *Prometheus) SetPushGateway(pushGatewayURL, metricsURL string, pushIntervalSeconds time.Duration) {
 	p.Ppg.PushGatewayURL = pushGatewayURL
 	p.Ppg.MetricsURL = metricsURL
@@ -158,13 +159,13 @@ func (p *Prometheus) SetPushGateway(pushGatewayURL, metricsURL string, pushInter
 	p.startPushTicker()
 }
 
-// SetPushGatewayJob job name, defaults to "gin"
+// SetPushGatewayJob job name, defaults to "gin".
 func (p *Prometheus) SetPushGatewayJob(j string) {
 	p.Ppg.Job = j
 }
 
 // SetListenAddress for exposing metrics on address. If not set, it will be exposed at the
-// same address of the gin engine that is being used
+// same address of the gin engine that is being used.
 func (p *Prometheus) SetListenAddress(address string) {
 	p.listenAddress = address
 	if p.listenAddress != "" {
@@ -181,7 +182,7 @@ func (p *Prometheus) SetListenAddressWithRouter(listenAddress string, r *gin.Eng
 	}
 }
 
-// SetMetricsPath set metrics paths
+// SetMetricsPath set metrics paths.
 func (p *Prometheus) SetMetricsPath(e *gin.Engine) {
 
 	if p.listenAddress != "" {
@@ -192,7 +193,7 @@ func (p *Prometheus) SetMetricsPath(e *gin.Engine) {
 	}
 }
 
-// SetMetricsPathWithAuth set metrics paths with authentication
+// SetMetricsPathWithAuth set metrics paths with authentication.
 func (p *Prometheus) SetMetricsPathWithAuth(e *gin.Engine, accounts gin.Accounts) {
 
 	if p.listenAddress != "" {
@@ -205,34 +206,43 @@ func (p *Prometheus) SetMetricsPathWithAuth(e *gin.Engine, accounts gin.Accounts
 }
 
 func (p *Prometheus) runServer() {
-	if p.listenAddress != "" {
-		go p.router.Run(p.listenAddress)
-	}
+	go p.router.Run(p.listenAddress)
 }
 
 func (p *Prometheus) getMetrics() []byte {
-	response, _ := http.Get(p.Ppg.MetricsURL)
+	response, err := http.Get(p.Ppg.MetricsURL)
+	if err != nil {
+		return nil
+	}
 
 	defer response.Body.Close()
-	body, _ := ioutil.ReadAll(response.Body)
 
+	body, _ := io.ReadAll(response.Body)
 	return body
 }
 
+var hostname, _ = os.Hostname()
+
 func (p *Prometheus) getPushGatewayURL() string {
-	h, _ := os.Hostname()
 	if p.Ppg.Job == "" {
 		p.Ppg.Job = "gin"
 	}
-	return p.Ppg.PushGatewayURL + "/metrics/job/" + p.Ppg.Job + "/instance/" + h
+	return p.Ppg.PushGatewayURL + "/metrics/job/" + p.Ppg.Job + "/instance/" + hostname
 }
 
 func (p *Prometheus) sendMetricsToPushGateway(metrics []byte) {
 	req, err := http.NewRequest("POST", p.getPushGatewayURL(), bytes.NewBuffer(metrics))
+	if err != nil {
+		return
+	}
+
 	client := &http.Client{}
-	if _, err = client.Do(req); err != nil {
+	resp, err := client.Do(req)
+	if err != nil {
 		fmt.Println("Error sending to push gateway error:", err.Error())
 	}
+
+	resp.Body.Close()
 }
 
 func (p *Prometheus) startPushTicker() {
@@ -244,7 +254,7 @@ func (p *Prometheus) startPushTicker() {
 	}()
 }
 
-// NewMetric associates prometheus.Collector based on Metric.Type
+// NewMetric associates prometheus.Collector based on Metric.Type.
 func NewMetric(m *Metric, subsystem string) prometheus.Collector {
 	var metric prometheus.Collector
 	switch m.Type {
@@ -321,20 +331,20 @@ func NewMetric(m *Metric, subsystem string) prometheus.Collector {
 }
 
 func (p *Prometheus) registerMetrics(subsystem string) {
-
 	for _, metricDef := range p.MetricsList {
 		metric := NewMetric(metricDef, subsystem)
 		if err := prometheus.Register(metric); err != nil {
 			fmt.Println("could not be registered in Prometheus,metricDef.Name:", metricDef.Name, "   error:", err.Error())
 		}
+
 		switch metricDef {
-		case reqCnt:
+		case reqCounter:
 			p.reqCnt = metric.(*prometheus.CounterVec)
-		case reqDur:
+		case reqDuration:
 			p.reqDur = metric.(*prometheus.HistogramVec)
-		case resSz:
+		case resSize:
 			p.resSz = metric.(prometheus.Summary)
-		case reqSz:
+		case reqSize:
 			p.reqSz = metric.(prometheus.Summary)
 		}
 		metricDef.MetricCollector = metric
@@ -353,7 +363,7 @@ func (p *Prometheus) UseWithAuth(e *gin.Engine, accounts gin.Accounts) {
 	p.SetMetricsPathWithAuth(e, accounts)
 }
 
-// HandlerFunc defines handler function for middleware
+// HandlerFunc defines handler function for middleware.
 func (p *Prometheus) HandlerFunc() gin.HandlerFunc {
 	return func(c *gin.Context) {
 		if c.Request.URL.Path == p.MetricsPath {
@@ -393,7 +403,7 @@ func prometheusHandler() gin.HandlerFunc {
 }
 
 func computeApproximateRequestSize(r *http.Request) int {
-	s := 0
+	var s int
 	if r.URL != nil {
 		s = len(r.URL.Path)
 	}


Test patch:
diff --git a/internal/tools/cron_task_test.go b/internal/tools/cron_task_test.go
new file mode 100644
index 000000000..2fcfba01b
--- /dev/null
+++ b/internal/tools/cron_task_test.go
@@ -0,0 +1,81 @@
+package tools
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/common/config"
+	"github.com/redis/go-redis/v9"
+	"github.com/robfig/cron/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDisLock(t *testing.T) {
+	rdb := redis.NewClient(&redis.Options{})
+	defer rdb.Close()
+
+	assert.Equal(t, true, netlock(rdb, "cron-1", 1*time.Second))
+
+	// if exists, get false
+	assert.Equal(t, false, netlock(rdb, "cron-1", 1*time.Second))
+
+	time.Sleep(2 * time.Second)
+
+	// wait for key on timeout, get true
+	assert.Equal(t, true, netlock(rdb, "cron-1", 2*time.Second))
+
+	// set different key
+	assert.Equal(t, true, netlock(rdb, "cron-2", 2*time.Second))
+}
+
+func TestCronWrapFunc(t *testing.T) {
+	rdb := redis.NewClient(&redis.Options{})
+	defer rdb.Close()
+
+	once := sync.Once{}
+	done := make(chan struct{}, 1)
+	cb := func() {
+		once.Do(func() {
+			close(done)
+		})
+	}
+
+	start := time.Now()
+	key := fmt.Sprintf("cron-%v", rand.Int31())
+	crontab := cron.New(cron.WithSeconds())
+	crontab.AddFunc("*/1 * * * * *", cronWrapFunc(rdb, key, cb))
+	crontab.Start()
+	<-done
+
+	dur := time.Since(start)
+	assert.LessOrEqual(t, dur.Seconds(), float64(2*time.Second))
+	crontab.Stop()
+}
+
+func TestCronWrapFuncWithNetlock(t *testing.T) {
+	config.Config.EnableCronLocker = true
+	rdb := redis.NewClient(&redis.Options{})
+	defer rdb.Close()
+
+	done := make(chan string, 10)
+
+	crontab := cron.New(cron.WithSeconds())
+
+	key := fmt.Sprintf("cron-%v", rand.Int31())
+	crontab.AddFunc("*/1 * * * * *", cronWrapFunc(rdb, key, func() {
+		done <- "host1"
+	}))
+	crontab.AddFunc("*/1 * * * * *", cronWrapFunc(rdb, key, func() {
+		done <- "host2"
+	}))
+	crontab.Start()
+
+	time.Sleep(12 * time.Second)
+	// the ttl of netlock is 5s, so expected value is 2.
+	assert.Equal(t, len(done), 2)
+
+	crontab.Stop()
+}