Fix push (#2559)

* fix:log

* fix: add log

* fix: del return

* fix: push config

* feat: add push err log and extend push wait time

* feat: group config

* feat: add log in write binary msg

* feat: Modify Prometheus data scraping ports and reserve port space.

* feat: change group rpc num

* feat: change log

* fix: remove quotation mark

config/openim-api.yml

@@ -8,6 +8,6 @@ prometheus:
   # Whether to enable prometheus
   enable: true
   # Prometheus listening ports, must match the number of api.ports
-  ports: [ 20113 ]
+  ports: [ 20502 ]
   # This address can be accessed via a browser
   grafanaURL: http://127.0.0.1:13000/

config/openim-msggateway.yml

@@ -8,7 +8,7 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20112 ]
+  ports: [ 20640 ]
 
 # IP address that the RPC/WebSocket service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
 listenIP: 0.0.0.0

config/openim-msgtransfer.yml

@@ -3,4 +3,4 @@ prometheus:
   enable: true
   # List of ports that Prometheus listens on; each port corresponds to an instance of monitoring. Ensure these are managed accordingly
   # Because four instances have been launched, four ports need to be specified
-  ports: [ 20108, 20109, 20110, 20111 ]
+  ports: [ 20600, 20601, 20602, 20603 ]

config/openim-push.yml

@@ -4,13 +4,13 @@ rpc:
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
-  ports: [ 10170 ]
+  ports: [ 10170, 10171, 10172, 10173 ]
 
 prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20107 ]
+  ports: [ 20670, 20671, 20672, 20673 ]
 
 maxConcurrentWorkers: 3
 #Use geTui for offline push notifications, or choose fcm or jpns; corresponding configuration settings must be specified.

config/openim-rpc-auth.yml

@@ -10,7 +10,7 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20106 ]
+  ports: [ 20660 ]
 
 tokenPolicy:
   # Token validity period, in days

config/openim-rpc-conversation.yml

@@ -10,4 +10,4 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20105 ]
+  ports: [ 20680 ]

config/openim-rpc-friend.yml

@@ -10,4 +10,4 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20104 ]
+  ports: [ 20620 ]

config/openim-rpc-group.yml

@@ -1,6 +1,6 @@
 rpc:
   # The IP address where this RPC service registers itself; if left blank, it defaults to the internal network IP
-  registerIP: 
+  registerIP:
   # IP address that the RPC service listens on; setting to 0.0.0.0 listens on both internal and external IPs. If left blank, it automatically uses the internal network IP
   listenIP: 0.0.0.0
   # List of ports that the RPC service listens on; configuring multiple ports will launch multiple instances. These must match the number of configured prometheus ports
@@ -10,7 +10,7 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20103 ]
+  ports: [ 20650 ]
 
 
 enableHistoryForNewMembers: true

config/openim-rpc-msg.yml

@@ -10,7 +10,7 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20102 ]
+  ports: [ 20630 ]
 
 
 # Does sending messages require friend verification

config/openim-rpc-third.yml

@@ -10,7 +10,7 @@ prometheus:
   # Enable or disable Prometheus monitoring
   enable: true
   # List of ports that Prometheus listens on; these must match the number of rpc.ports to ensure correct monitoring setup
-  ports: [ 20101 ]
+  ports: [ 20690 ]
 
 
 object:

config/openim-rpc-user.yml

@@ -10,7 +10,7 @@ prometheus:
   # Whether to enable prometheus
   enable: true
   # Prometheus listening ports, must be consistent with the number of rpc.ports
-  ports: [ 20100 ]
+  ports: [ 20610 ]
 
 
 

config/prometheus.yml

@@ -19,65 +19,65 @@ rule_files:
 # A scrape configuration containing exactly one endpoint to scrape:
 # Here it's Prometheus itself.
 scrape_configs:
-  # The job name is added as a label job=job_name" to any timeseries scraped from this config.
+  # The job name is added as a label "job=job_name" to any timeseries scraped from this config.
   # Monitored information captured by prometheus
 
   # prometheus fetches application services
   - job_name: node_exporter
     static_configs:
-      - targets: [ internal_ip:20114 ]
+      - targets: [ internal_ip:20500 ]
   - job_name: openimserver-openim-api
     static_configs:
-      - targets: [ internal_ip:20113 ]
+      - targets: [ internal_ip:20502 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-msggateway
     static_configs:
-      - targets: [ internal_ip:20112 ]
+      - targets: [ internal_ip:20640 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-msgtransfer
     static_configs:
-      - targets: [ internal_ip:20111, internal_ip:20110, internal_ip:20109, internal_ip:20108 ]
+      - targets: [ internal_ip:20600, internal_ip:20601, internal_ip:20602, internal_ip:20603 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-push
     static_configs:
-      - targets: [ internal_ip:20107 ]
+      - targets: [ internal_ip:20670, internal_ip:20671, internal_ip:20672, internal_ip:20673]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-auth
     static_configs:
-      - targets: [ internal_ip:20106 ]
+      - targets: [ internal_ip:20600 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-conversation
     static_configs:
-      - targets: [ internal_ip:20105 ]
+      - targets: [ internal_ip:20680 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-friend
     static_configs:
-      - targets: [ internal_ip:20104 ]
+      - targets: [ internal_ip:20620 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-group
     static_configs:
-      - targets: [ internal_ip:20103 ]
+      - targets: [ internal_ip:20650 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-msg
     static_configs:
-      - targets: [ internal_ip:20102 ]
+      - targets: [ internal_ip:20630 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-third
     static_configs:
-      - targets: [ internal_ip:20101 ]
+      - targets: [ internal_ip:20690 ]
         labels:
           namespace: default
   - job_name: openimserver-openim-rpc-user
     static_configs:
-      - targets: [ internal_ip:20100 ]
+      - targets: [ internal_ip:20610 ]
         labels:
           namespace: default

internal/msggateway/client.go

@@ -271,11 +271,13 @@ func (c *Client) replyMessage(ctx context.Context, binaryReq *Req, err error, re
 		ErrMsg:        errResp.ErrMsg,
 		Data:          resp,
 	}
+	t := time.Now()
 	log.ZDebug(ctx, "gateway reply message", "resp", mReply.String())
 	err = c.writeBinaryMsg(mReply)
 	if err != nil {
 		log.ZWarn(ctx, "wireBinaryMsg replyMessage", err, "resp", mReply.String())
 	}
+	log.ZDebug(ctx, "wireBinaryMsg end", "time cost", time.Since(t))
 
 	if binaryReq.ReqIdentifier == WsLogoutMsg {
 		return errs.New("user logout", "operationID", binaryReq.OperationID).Wrap()

internal/push/push_handler.go

@@ -91,9 +91,9 @@ func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) {
 	}
 	sec := msgFromMQ.MsgData.SendTime / 1000
 	nowSec := timeutil.GetCurrentTimestampBySecond()
-	log.ZDebug(ctx, "push msg", "msg", pbData.String(), "sec", sec, "nowSec", nowSec)
+
 	if nowSec-sec > 10 {
-		return
+		log.ZWarn(ctx, "long time push msg", nil, "msg", pbData.String(), "sec", sec, "nowSec", nowSec, "nowSec-sec", nowSec-sec)
 	}
 	var err error
 	switch msgFromMQ.MsgData.SessionType {

pkg/rpcclient/msg.go

@@ -355,7 +355,9 @@ func (s *NotificationSender) send(ctx context.Context, sendID, recvID string, co
 }
 
 func (s *NotificationSender) NotificationWithSessionType(ctx context.Context, sendID, recvID string, contentType, sessionType int32, m proto.Message, opts ...NotificationOptions) {
-	s.queue.Push(func() { s.send(ctx, sendID, recvID, contentType, sessionType, m, opts...) })
+	if err := s.queue.Push(func() { s.send(ctx, sendID, recvID, contentType, sessionType, m, opts...) }); err != nil {
+		log.ZWarn(ctx, "Push to queue failed", err, "sendID", sendID, "recvID", recvID, "msg", jsonutil.StructToJsonString(m))
+	}
 }
 
 func (s *NotificationSender) Notification(ctx context.Context, sendID, recvID string, contentType int32, m proto.Message, opts ...NotificationOptions) {

start-config.yml

@@ -3,7 +3,7 @@ serviceBinaries:
   openim-crontask: 1
   openim-rpc-user: 1
   openim-msggateway: 1
-  openim-push: 1
+  openim-push: 4
   openim-msgtransfer: 4
   openim-rpc-conversation: 1
   openim-rpc-auth: 1