mirror of
https://github.com/Sagit-chu/flvx.git
synced 2026-10-07 02:06:38 +08:00
chore: optimize agent-panel metrics communication (#352)
This commit is contained in:
@@ -1234,6 +1234,21 @@ func nullableNullInt64(v sql.NullInt64) interface{} {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// flowCryptoCache caches AES crypto instances by secret to avoid per-request SHA256+GCM init.
|
||||||
|
var flowCryptoCache sync.Map
|
||||||
|
|
||||||
|
func getOrCreateFlowCrypto(secret string) *security.AESCrypto {
|
||||||
|
if v, ok := flowCryptoCache.Load(secret); ok {
|
||||||
|
return v.(*security.AESCrypto)
|
||||||
|
}
|
||||||
|
c, err := security.NewAESCrypto(secret)
|
||||||
|
if err != nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
flowCryptoCache.Store(secret, c)
|
||||||
|
return c
|
||||||
|
}
|
||||||
|
|
||||||
func readAndDecryptFlowBody(body io.ReadCloser, secret string) (string, error) {
|
func readAndDecryptFlowBody(body io.ReadCloser, secret string) (string, error) {
|
||||||
defer body.Close()
|
defer body.Close()
|
||||||
raw, err := io.ReadAll(body)
|
raw, err := io.ReadAll(body)
|
||||||
@@ -1254,8 +1269,8 @@ func readAndDecryptFlowBody(body io.ReadCloser, secret string) (string, error) {
|
|||||||
return text, nil
|
return text, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
crypto, err := security.NewAESCrypto(secret)
|
crypto := getOrCreateFlowCrypto(secret)
|
||||||
if err != nil {
|
if crypto == nil {
|
||||||
return text, nil
|
return text, nil
|
||||||
}
|
}
|
||||||
plain, err := crypto.Decrypt(wrap.Data)
|
plain, err := crypto.Decrypt(wrap.Data)
|
||||||
|
|||||||
@@ -39,6 +39,7 @@ type nodeSession struct {
|
|||||||
nodeID int64
|
nodeID int64
|
||||||
secret string
|
secret string
|
||||||
conn *connWrap
|
conn *connWrap
|
||||||
|
crypto *security.AESCrypto // 缓存的 AES 加密器,避免每条消息重建
|
||||||
}
|
}
|
||||||
|
|
||||||
type commandResponse struct {
|
type commandResponse struct {
|
||||||
@@ -211,7 +212,12 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
|
|||||||
_ = old.conn.conn.Close()
|
_ = old.conn.conn.Close()
|
||||||
delete(s.byConn, old.conn.conn)
|
delete(s.byConn, old.conn.conn)
|
||||||
}
|
}
|
||||||
ns := &nodeSession{nodeID: nodeID, secret: secret, conn: cw}
|
// 初始化 AES 加密器并缓存(仅创建一次)
|
||||||
|
var nodeCrypto *security.AESCrypto
|
||||||
|
if strings.TrimSpace(secret) != "" {
|
||||||
|
nodeCrypto, _ = security.NewAESCrypto(secret)
|
||||||
|
}
|
||||||
|
ns := &nodeSession{nodeID: nodeID, secret: secret, conn: cw, crypto: nodeCrypto}
|
||||||
s.nodes[nodeID] = ns
|
s.nodes[nodeID] = ns
|
||||||
s.byConn[conn] = ns
|
s.byConn[conn] = ns
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
@@ -251,7 +257,7 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
msg := decryptIfNeeded(payload, secret)
|
msg := decryptIfNeeded(payload, ns.crypto, secret)
|
||||||
s.tryResolvePending(nodeID, msg)
|
s.tryResolvePending(nodeID, msg)
|
||||||
|
|
||||||
var parsed struct {
|
var parsed struct {
|
||||||
@@ -259,6 +265,26 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
|
|||||||
}
|
}
|
||||||
if json.Unmarshal([]byte(msg), &parsed) == nil && parsed.Type != "" {
|
if json.Unmarshal([]byte(msg), &parsed) == nil && parsed.Type != "" {
|
||||||
switch parsed.Type {
|
switch parsed.Type {
|
||||||
|
case "metric":
|
||||||
|
// Agent 新版指标消息:{type:"metric", data:{...}}
|
||||||
|
var envelope struct {
|
||||||
|
Data json.RawMessage `json:"data"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal([]byte(msg), &envelope); err == nil && len(envelope.Data) > 0 {
|
||||||
|
// 解析 SystemInfo 并调用 hook
|
||||||
|
var sysInfo SystemInfo
|
||||||
|
if json.Unmarshal(envelope.Data, &sysInfo) == nil {
|
||||||
|
s.mu.RLock()
|
||||||
|
onMetric := s.onNodeMetric
|
||||||
|
s.mu.RUnlock()
|
||||||
|
if onMetric != nil {
|
||||||
|
go onMetric(nodeID, sysInfo)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// 广播内层 data 给前端(保持平坦结构兼容性)
|
||||||
|
s.broadcastTyped(nodeID, "metric", string(envelope.Data))
|
||||||
|
}
|
||||||
|
continue
|
||||||
case "UpgradeProgress":
|
case "UpgradeProgress":
|
||||||
s.broadcastTyped(nodeID, "upgrade_progress", msg)
|
s.broadcastTyped(nodeID, "upgrade_progress", msg)
|
||||||
continue
|
continue
|
||||||
@@ -270,6 +296,7 @@ func (s *Server) handleNode(w http.ResponseWriter, r *http.Request, nodeID int64
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 兼容旧版 Agent:无 type 字段的系统信息消息
|
||||||
if looksLikeSystemInfoMessage(msg) {
|
if looksLikeSystemInfoMessage(msg) {
|
||||||
var sysInfo SystemInfo
|
var sysInfo SystemInfo
|
||||||
if err := json.Unmarshal([]byte(msg), &sysInfo); err == nil {
|
if err := json.Unmarshal([]byte(msg), &sysInfo); err == nil {
|
||||||
@@ -372,13 +399,8 @@ func (s *Server) SendCommand(nodeID int64, cmdType string, data interface{}, tim
|
|||||||
}
|
}
|
||||||
|
|
||||||
messageData := rawCmd
|
messageData := rawCmd
|
||||||
if strings.TrimSpace(ns.secret) != "" {
|
if ns.crypto != nil {
|
||||||
crypto, err := security.NewAESCrypto(ns.secret)
|
encrypted, err := ns.crypto.Encrypt(rawCmd)
|
||||||
if err != nil {
|
|
||||||
cleanup()
|
|
||||||
return CommandResult{}, err
|
|
||||||
}
|
|
||||||
encrypted, err := crypto.Encrypt(rawCmd)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
cleanup()
|
cleanup()
|
||||||
return CommandResult{}, err
|
return CommandResult{}, err
|
||||||
@@ -428,6 +450,11 @@ func (s *Server) tryResolvePending(nodeID int64, message string) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 快速短路:指标消息永远不含 requestId,跳过完整 JSON 解析
|
||||||
|
if !strings.Contains(message, "\"requestId\"") {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
var resp commandResponse
|
var resp commandResponse
|
||||||
if err := json.Unmarshal([]byte(message), &resp); err != nil {
|
if err := json.Unmarshal([]byte(message), &resp); err != nil {
|
||||||
return
|
return
|
||||||
@@ -545,18 +572,22 @@ func (s *Server) broadcastToAdmins(message string) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func decryptIfNeeded(payload []byte, secret string) string {
|
func decryptIfNeeded(payload []byte, crypto *security.AESCrypto, secret string) string {
|
||||||
text := string(payload)
|
text := string(payload)
|
||||||
var wrap encryptedMessage
|
var wrap encryptedMessage
|
||||||
if err := json.Unmarshal(payload, &wrap); err != nil || !wrap.Encrypted || strings.TrimSpace(wrap.Data) == "" {
|
if err := json.Unmarshal(payload, &wrap); err != nil || !wrap.Encrypted || strings.TrimSpace(wrap.Data) == "" {
|
||||||
return text
|
return text
|
||||||
}
|
}
|
||||||
|
|
||||||
crypto, err := security.NewAESCrypto(secret)
|
// 优先使用缓存的 crypto 实例
|
||||||
if err != nil {
|
c := crypto
|
||||||
|
if c == nil && strings.TrimSpace(secret) != "" {
|
||||||
|
c, _ = security.NewAESCrypto(secret)
|
||||||
|
}
|
||||||
|
if c == nil {
|
||||||
return text
|
return text
|
||||||
}
|
}
|
||||||
plain, err := crypto.Decrypt(wrap.Data)
|
plain, err := c.Decrypt(wrap.Data)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return text
|
return text
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
|
"math/rand"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/url"
|
"net/url"
|
||||||
@@ -146,6 +147,9 @@ type ServiceMonitorCheckResult struct {
|
|||||||
const (
|
const (
|
||||||
reporterReadWait = 60 * time.Second
|
reporterReadWait = 60 * time.Second
|
||||||
reporterWriteWait = 5 * time.Second
|
reporterWriteWait = 5 * time.Second
|
||||||
|
wsPingInterval = 20 * time.Second // 独立 WebSocket ping 间隔
|
||||||
|
initialBackoff = 2 * time.Second // 重连初始退避
|
||||||
|
maxBackoff = 2 * time.Minute // 重连最大退避
|
||||||
)
|
)
|
||||||
|
|
||||||
type WebSocketReporter struct {
|
type WebSocketReporter struct {
|
||||||
@@ -155,15 +159,15 @@ type WebSocketReporter struct {
|
|||||||
version string // 保存版本号
|
version string // 保存版本号
|
||||||
preferredWSScheme string
|
preferredWSScheme string
|
||||||
conn *websocket.Conn
|
conn *websocket.Conn
|
||||||
reconnectTime time.Duration
|
curBackoff time.Duration // 当前重连退避间隔
|
||||||
pingInterval time.Duration
|
pingInterval time.Duration
|
||||||
configInterval time.Duration
|
configInterval time.Duration
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
connected bool
|
connected bool
|
||||||
connecting bool // 新增:正在连接状态
|
connecting bool // 正在连接状态
|
||||||
connMutex sync.Mutex // 新增:连接状态锁
|
connMutex sync.Mutex // 连接状态锁
|
||||||
aesCrypto *crypto.AESCrypto // 新增:AES加密器
|
aesCrypto *crypto.AESCrypto // AES加密器
|
||||||
}
|
}
|
||||||
|
|
||||||
var wsDial = func(dialer *websocket.Dialer, rawURL string) (*websocket.Conn, *http.Response, error) {
|
var wsDial = func(dialer *websocket.Dialer, rawURL string) (*websocket.Conn, *http.Response, error) {
|
||||||
@@ -185,7 +189,7 @@ func NewWebSocketReporter(serverURL string, secret string) *WebSocketReporter {
|
|||||||
|
|
||||||
return &WebSocketReporter{
|
return &WebSocketReporter{
|
||||||
url: serverURL,
|
url: serverURL,
|
||||||
reconnectTime: 5 * time.Second, // 重连间隔
|
curBackoff: initialBackoff, // 当前退避间隔
|
||||||
pingInterval: 5 * time.Second, // 指标上报间隔
|
pingInterval: 5 * time.Second, // 指标上报间隔
|
||||||
configInterval: 10 * time.Minute, // 配置上报间隔
|
configInterval: 10 * time.Minute, // 配置上报间隔
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
@@ -204,10 +208,17 @@ func (w *WebSocketReporter) Start() {
|
|||||||
// Stop 停止WebSocket报告器
|
// Stop 停止WebSocket报告器
|
||||||
func (w *WebSocketReporter) Stop() {
|
func (w *WebSocketReporter) Stop() {
|
||||||
w.cancel()
|
w.cancel()
|
||||||
|
w.connMutex.Lock()
|
||||||
if w.conn != nil {
|
if w.conn != nil {
|
||||||
w.conn.Close()
|
w.conn.Close()
|
||||||
}
|
}
|
||||||
|
w.connMutex.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// backoffWithJitter 返回带随机抖动的退避时间(±25%)
|
||||||
|
func backoffWithJitter(base time.Duration) time.Duration {
|
||||||
|
jitter := time.Duration(float64(base) * (0.75 + rand.Float64()*0.5))
|
||||||
|
return jitter
|
||||||
}
|
}
|
||||||
|
|
||||||
// run 主运行循环
|
// run 主运行循环
|
||||||
@@ -224,23 +235,32 @@ func (w *WebSocketReporter) run() {
|
|||||||
|
|
||||||
if needConnect {
|
if needConnect {
|
||||||
if err := w.connect(); err != nil {
|
if err := w.connect(); err != nil {
|
||||||
fmt.Printf("❌ WebSocket连接失败: %v,%v后重试\n", err, w.reconnectTime)
|
wait := backoffWithJitter(w.curBackoff)
|
||||||
|
fmt.Printf("❌ WebSocket连接失败: %v,%v后重试\n", err, wait)
|
||||||
|
// 指数退避:翻倍当前退避间隔,上限 maxBackoff
|
||||||
|
w.curBackoff *= 2
|
||||||
|
if w.curBackoff > maxBackoff {
|
||||||
|
w.curBackoff = maxBackoff
|
||||||
|
}
|
||||||
select {
|
select {
|
||||||
case <-time.After(w.reconnectTime):
|
case <-time.After(wait):
|
||||||
continue
|
continue
|
||||||
case <-w.ctx.Done():
|
case <-w.ctx.Done():
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
// 连接成功:重置退避
|
||||||
|
w.curBackoff = initialBackoff
|
||||||
}
|
}
|
||||||
|
|
||||||
// 连接成功,开始发送消息
|
// 连接成功,开始发送消息
|
||||||
if w.connected {
|
if w.connected {
|
||||||
w.handleConnection()
|
w.handleConnection()
|
||||||
} else {
|
} else {
|
||||||
|
wait := backoffWithJitter(w.curBackoff)
|
||||||
// 如果连接失败,等待重试
|
// 如果连接失败,等待重试
|
||||||
select {
|
select {
|
||||||
case <-time.After(w.reconnectTime):
|
case <-time.After(wait):
|
||||||
continue
|
continue
|
||||||
case <-w.ctx.Done():
|
case <-w.ctx.Done():
|
||||||
return
|
return
|
||||||
@@ -473,15 +493,34 @@ func (w *WebSocketReporter) handleConnection() {
|
|||||||
// 启动消息接收goroutine
|
// 启动消息接收goroutine
|
||||||
go w.receiveMessages()
|
go w.receiveMessages()
|
||||||
|
|
||||||
// 主发送循环
|
// 指标上报 ticker
|
||||||
ticker := time.NewTicker(w.pingInterval)
|
metricTicker := time.NewTicker(w.pingInterval)
|
||||||
defer ticker.Stop()
|
defer metricTicker.Stop()
|
||||||
|
|
||||||
|
// 独立 WebSocket keepalive ping ticker
|
||||||
|
pingTicker := time.NewTicker(wsPingInterval)
|
||||||
|
defer pingTicker.Stop()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-w.ctx.Done():
|
case <-w.ctx.Done():
|
||||||
return
|
return
|
||||||
case <-ticker.C:
|
|
||||||
|
case <-pingTicker.C:
|
||||||
|
// 发送 WebSocket ping 保活,独立于指标上报
|
||||||
|
w.connMutex.Lock()
|
||||||
|
conn := w.conn
|
||||||
|
isConnected := w.connected
|
||||||
|
w.connMutex.Unlock()
|
||||||
|
if !isConnected || conn == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(reporterWriteWait)); err != nil {
|
||||||
|
fmt.Printf("❌ 发送WebSocket ping失败: %v,准备重连\n", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
case <-metricTicker.C:
|
||||||
// 检查连接状态
|
// 检查连接状态
|
||||||
w.connMutex.Lock()
|
w.connMutex.Lock()
|
||||||
isConnected := w.connected
|
isConnected := w.connected
|
||||||
@@ -548,6 +587,37 @@ func (w *WebSocketReporter) collectSystemInfo() SystemInfo {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// encryptPayload 加密 JSON 数据,返回加密后的消息字节(若加密失败则回退到原始数据)
|
||||||
|
func (w *WebSocketReporter) encryptPayload(jsonData []byte) []byte {
|
||||||
|
if w.aesCrypto == nil {
|
||||||
|
return jsonData
|
||||||
|
}
|
||||||
|
|
||||||
|
encryptedData, err := w.aesCrypto.Encrypt(jsonData)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Printf("⚠️ 加密失败,发送原始数据: %v\n", err)
|
||||||
|
return jsonData
|
||||||
|
}
|
||||||
|
|
||||||
|
encryptedMessage := map[string]interface{}{
|
||||||
|
"encrypted": true,
|
||||||
|
"data": encryptedData,
|
||||||
|
"timestamp": time.Now().Unix(),
|
||||||
|
}
|
||||||
|
messageData, err := json.Marshal(encryptedMessage)
|
||||||
|
if err != nil {
|
||||||
|
fmt.Printf("⚠️ 序列化加密消息失败,发送原始数据: %v\n", err)
|
||||||
|
return jsonData
|
||||||
|
}
|
||||||
|
return messageData
|
||||||
|
}
|
||||||
|
|
||||||
|
// metricEnvelope wraps SystemInfo with a type field for fast identification on the panel side.
|
||||||
|
type metricEnvelope struct {
|
||||||
|
Type string `json:"type"`
|
||||||
|
Data SystemInfo `json:"data"`
|
||||||
|
}
|
||||||
|
|
||||||
// sendSystemInfo 发送系统信息
|
// sendSystemInfo 发送系统信息
|
||||||
func (w *WebSocketReporter) sendSystemInfo(sysInfo SystemInfo) error {
|
func (w *WebSocketReporter) sendSystemInfo(sysInfo SystemInfo) error {
|
||||||
w.connMutex.Lock()
|
w.connMutex.Lock()
|
||||||
@@ -557,42 +627,19 @@ func (w *WebSocketReporter) sendSystemInfo(sysInfo SystemInfo) error {
|
|||||||
return fmt.Errorf("连接未建立")
|
return fmt.Errorf("连接未建立")
|
||||||
}
|
}
|
||||||
|
|
||||||
// 转换为JSON
|
// 使用 type:"metric" 信封包装,Panel 可通过 type 字段直接识别指标消息
|
||||||
jsonData, err := json.Marshal(sysInfo)
|
envelope := metricEnvelope{Type: "metric", Data: sysInfo}
|
||||||
|
jsonData, err := json.Marshal(envelope)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("序列化系统信息失败: %v", err)
|
return fmt.Errorf("序列化系统信息失败: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
var messageData []byte
|
messageData := w.encryptPayload(jsonData)
|
||||||
|
|
||||||
// 如果有加密器,则加密数据
|
|
||||||
if w.aesCrypto != nil {
|
|
||||||
encryptedData, err := w.aesCrypto.Encrypt(jsonData)
|
|
||||||
if err != nil {
|
|
||||||
fmt.Printf("⚠️ 加密失败,发送原始数据: %v\n", err)
|
|
||||||
messageData = jsonData
|
|
||||||
} else {
|
|
||||||
// 创建加密消息包装器
|
|
||||||
encryptedMessage := map[string]interface{}{
|
|
||||||
"encrypted": true,
|
|
||||||
"data": encryptedData,
|
|
||||||
"timestamp": time.Now().Unix(),
|
|
||||||
}
|
|
||||||
messageData, err = json.Marshal(encryptedMessage)
|
|
||||||
if err != nil {
|
|
||||||
fmt.Printf("⚠️ 序列化加密消息失败,发送原始数据: %v\n", err)
|
|
||||||
messageData = jsonData
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
messageData = jsonData
|
|
||||||
}
|
|
||||||
|
|
||||||
// 设置写入超时
|
|
||||||
w.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
|
w.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
|
||||||
|
|
||||||
if err := w.conn.WriteMessage(websocket.TextMessage, messageData); err != nil {
|
if err := w.conn.WriteMessage(websocket.TextMessage, messageData); err != nil {
|
||||||
w.connected = false // 标记连接已断开
|
w.connected = false
|
||||||
return fmt.Errorf("写入消息失败: %v", err)
|
return fmt.Errorf("写入消息失败: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -601,23 +648,19 @@ func (w *WebSocketReporter) sendSystemInfo(sysInfo SystemInfo) error {
|
|||||||
|
|
||||||
// receiveMessages 接收服务端发送的消息
|
// receiveMessages 接收服务端发送的消息
|
||||||
func (w *WebSocketReporter) receiveMessages() {
|
func (w *WebSocketReporter) receiveMessages() {
|
||||||
|
// 获取连接引用一次即可,连接生命周期由 handleConnection 管理
|
||||||
|
w.connMutex.Lock()
|
||||||
|
conn := w.conn
|
||||||
|
w.connMutex.Unlock()
|
||||||
|
if conn == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-w.ctx.Done():
|
case <-w.ctx.Done():
|
||||||
return
|
return
|
||||||
default:
|
default:
|
||||||
w.connMutex.Lock()
|
|
||||||
conn := w.conn
|
|
||||||
connected := w.connected
|
|
||||||
w.connMutex.Unlock()
|
|
||||||
|
|
||||||
if conn == nil || !connected {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// 设置读取超时
|
|
||||||
conn.SetReadDeadline(time.Now().Add(reporterReadWait))
|
|
||||||
|
|
||||||
messageType, message, err := conn.ReadMessage()
|
messageType, message, err := conn.ReadMessage()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
|
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
|
||||||
@@ -705,12 +748,8 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
|
|||||||
}
|
}
|
||||||
|
|
||||||
if cmdMsg.Type != "call" {
|
if cmdMsg.Type != "call" {
|
||||||
// 其他状态变更命令保持同步,确保顺序执行
|
// 所有命令统一异步执行,避免阻塞消息接收循环
|
||||||
if cmdMsg.Type == "TcpPing" || cmdMsg.Type == "ServiceMonitorCheck" || cmdMsg.Type == "UpgradeAgent" || cmdMsg.Type == "RollbackAgent" {
|
go w.routeCommand(cmdMsg)
|
||||||
go w.routeCommand(cmdMsg)
|
|
||||||
} else {
|
|
||||||
w.routeCommand(cmdMsg)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// 处理普通消息
|
// 处理普通消息
|
||||||
@@ -721,12 +760,8 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if cmdMsg.Type != "call" {
|
if cmdMsg.Type != "call" {
|
||||||
// 其他状态变更命令保持同步,确保顺序执行
|
// 所有命令统一异步执行,避免阻塞消息接收循环
|
||||||
if cmdMsg.Type == "TcpPing" || cmdMsg.Type == "ServiceMonitorCheck" || cmdMsg.Type == "UpgradeAgent" || cmdMsg.Type == "RollbackAgent" {
|
go w.routeCommand(cmdMsg)
|
||||||
go w.routeCommand(cmdMsg)
|
|
||||||
} else {
|
|
||||||
w.routeCommand(cmdMsg)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1400,30 +1435,7 @@ func (w *WebSocketReporter) sendResponse(response CommandResponse) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
var messageData []byte
|
messageData := w.encryptPayload(jsonData)
|
||||||
|
|
||||||
// 如果有加密器,则加密数据
|
|
||||||
if w.aesCrypto != nil {
|
|
||||||
encryptedData, err := w.aesCrypto.Encrypt(jsonData)
|
|
||||||
if err != nil {
|
|
||||||
fmt.Printf("⚠️ 加密响应失败,发送原始数据: %v\n", err)
|
|
||||||
messageData = jsonData
|
|
||||||
} else {
|
|
||||||
// 创建加密消息包装器
|
|
||||||
encryptedMessage := map[string]interface{}{
|
|
||||||
"encrypted": true,
|
|
||||||
"data": encryptedData,
|
|
||||||
"timestamp": time.Now().Unix(),
|
|
||||||
}
|
|
||||||
messageData, err = json.Marshal(encryptedMessage)
|
|
||||||
if err != nil {
|
|
||||||
fmt.Printf("⚠️ 序列化加密响应失败,发送原始数据: %v\n", err)
|
|
||||||
messageData = jsonData
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
messageData = jsonData
|
|
||||||
}
|
|
||||||
|
|
||||||
// 检查消息大小,如果超过10MB则记录警告
|
// 检查消息大小,如果超过10MB则记录警告
|
||||||
if len(messageData) > 10*1024*1024 {
|
if len(messageData) > 10*1024*1024 {
|
||||||
|
|||||||
@@ -0,0 +1,147 @@
|
|||||||
|
# Agent-Panel 通信优化:提升稳定性与效率
|
||||||
|
|
||||||
|
## 背景
|
||||||
|
|
||||||
|
Agent(`go-gost/x/socket/websocket_reporter.go`)与 Panel(`go-backend/internal/ws/server.go`)之间通过 WebSocket 进行实时通信,包括指标上报(每 5s)、命令下发/响应、和流量上报(HTTP)。经过代码审查,以下是发现的问题和优化建议。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 发现的问题
|
||||||
|
|
||||||
|
### 1. Keepalive 时序不匹配 —— 导致误断连
|
||||||
|
|
||||||
|
| 参数 | Agent 侧 | Panel 侧 |
|
||||||
|
|------|----------|----------|
|
||||||
|
| Read deadline | `reporterReadWait` = 60s | `wsPongWait` = 45s |
|
||||||
|
| Ping 发送间隔 | 无主动 ping(靠指标数据 5s 续命) | `wsPingPeriod` = 15s |
|
||||||
|
| Write timeout | `reporterWriteWait` = 5s | `wsWriteWait` = 5s |
|
||||||
|
|
||||||
|
**问题**:Panel 每 15s 发 ping,Agent read deadline 60s,但 Panel pong deadline 只有 45s。如果 Agent 的指标消息被延迟(网络抖动),Panel 可能因 pong 超时而关闭连接。两侧的超时参数缺乏协调设计。
|
||||||
|
|
||||||
|
### 2. 固定重连间隔 —— 无退避策略
|
||||||
|
|
||||||
|
Agent 断线后以固定 5s 间隔重试(`reconnectTime = 5 * time.Second`),在 Panel 长时间不可用(升级、网络故障)的情况下,会产生大量无用连接尝试。
|
||||||
|
|
||||||
|
### 3. Panel 侧每次解密都重建 AES 加密器
|
||||||
|
|
||||||
|
`ws/server.go` 的 `decryptIfNeeded()` 和 `SendCommand()` 每次调用都 `security.NewAESCrypto(secret)` 重新创建 cipher(SHA256 + AES-GCM 初始化),对于高频指标消息(5s/次 × N 节点),有不必要的 CPU 开销。
|
||||||
|
|
||||||
|
### 4. 指标消息使用 JSON Text 格式传输
|
||||||
|
|
||||||
|
每 5s 发送一次包含 13 个字段的 SystemInfo JSON,加密后还需 base64 编码,一条消息约 300-500 bytes(加密后约 700 bytes)。对于大量节点场景,存在优化空间。
|
||||||
|
|
||||||
|
### 5. `receiveMessages` 紧循环中有频繁锁竞争
|
||||||
|
|
||||||
|
`receiveMessages()` 在每次 `ReadMessage()` 前都要 `Lock/Unlock connMutex` 检查连接状态,但 `ReadMessage` 本身是阻塞的,实际不需要在循环外检查。
|
||||||
|
|
||||||
|
### 6. 状态变更命令阻塞读消息循环
|
||||||
|
|
||||||
|
`routeCommand` 中的 Service/Chain/Limiter CRUD 命令是同步执行的,包括 `saveConfig()` 文件写入。执行期间会阻塞 `receiveMessages` 的读取循环。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 推荐的优化方案(按优先级排列)
|
||||||
|
|
||||||
|
### P0 — 高收益、低风险
|
||||||
|
|
||||||
|
#### 优化 1:协调 Keepalive 参数
|
||||||
|
|
||||||
|
**文件**:`websocket_reporter.go`
|
||||||
|
|
||||||
|
- Agent 增加独立的 WebSocket ping 发送(每 20s),不依赖指标数据来维持连接
|
||||||
|
- 统一 read deadline 设置,确保两侧 read timeout > 2×ping interval
|
||||||
|
|
||||||
|
#### 优化 2:指数退避重连
|
||||||
|
|
||||||
|
**文件**:`websocket_reporter.go`
|
||||||
|
|
||||||
|
- 初始间隔 2s,按指数退避增长至最大 2 分钟
|
||||||
|
- 连接成功后立即重置退避
|
||||||
|
- 增加随机抖动(jitter)避免大量 Agent 同时重连
|
||||||
|
|
||||||
|
#### 优化 3:Panel 侧缓存 AES 加密器
|
||||||
|
|
||||||
|
**文件**:`ws/server.go`
|
||||||
|
|
||||||
|
- 将 `AESCrypto` 实例缓存在 `nodeSession` 中,避免每条消息重建
|
||||||
|
- `SendCommand` 复用缓存实例
|
||||||
|
|
||||||
|
### P1 — 中等收益
|
||||||
|
|
||||||
|
#### 优化 4:减少 `receiveMessages` 锁竞争
|
||||||
|
|
||||||
|
**文件**:`websocket_reporter.go`
|
||||||
|
|
||||||
|
- 将连接状态检查移到循环外,只在出错/关闭时通过 channel 通知退出
|
||||||
|
- 用 `context.WithCancel` 代替锁检查 `connected` flag 来控制生命周期
|
||||||
|
|
||||||
|
#### 优化 5:异步化状态变更命令处理
|
||||||
|
|
||||||
|
**文件**:`websocket_reporter.go`
|
||||||
|
|
||||||
|
- 所有命令统一异步执行(通过 goroutine + response channel),避免阻塞 readLoop
|
||||||
|
- 当前只有 TcpPing/ServiceMonitorCheck/UpgradeAgent/RollbackAgent 是异步的
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 具体代码变更
|
||||||
|
|
||||||
|
### Agent 侧 (`go-gost/x/socket`)
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
#### [MODIFY] [websocket_reporter.go](file:///Users/sagit/Documents/github/flvx/go-gost/x/socket/websocket_reporter.go)
|
||||||
|
|
||||||
|
1. **指数退避重连**:将 `reconnectTime` 从固定 `5s` 改为动态退避字段,增加 `curBackoff/maxBackoff` 字段
|
||||||
|
2. **独立 Ping 发送**:在 `handleConnection()` 中增加 WebSocket ping ticker(20s),独立于指标上报
|
||||||
|
3. **减少锁竞争**:`receiveMessages` 中只在循环入口检查一次连接,此后靠 `ReadMessage` 的 error 退出
|
||||||
|
4. **统一命令异步化**:所有 `routeCommand` 调用统一使用 goroutine
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### Panel 侧 (`go-backend/internal/ws`)
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
#### [MODIFY] [server.go](file:///Users/sagit/Documents/github/flvx/go-backend/internal/ws/server.go)
|
||||||
|
|
||||||
|
1. **缓存 AES 加密器**:在 `nodeSession` 中增加 `crypto *security.AESCrypto` 字段,节点连接时初始化
|
||||||
|
2. **`decryptIfNeeded` 接收 crypto 参数**而非 secret 字符串
|
||||||
|
3. **`SendCommand` 使用缓存 crypto** 实例
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Verification Plan
|
||||||
|
|
||||||
|
### Automated Tests
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 运行现有 agent 侧单元测试(验证不回归)
|
||||||
|
(cd go-gost/x && go test ./socket/... -v -count=1)
|
||||||
|
|
||||||
|
# 运行现有流量上报测试
|
||||||
|
(cd go-gost/x && go test ./service/... -v -count=1)
|
||||||
|
|
||||||
|
# 运行 panel 侧全部测试
|
||||||
|
(cd go-backend && go test ./... -count=1)
|
||||||
|
```
|
||||||
|
|
||||||
|
### Manual Verification
|
||||||
|
|
||||||
|
> [!IMPORTANT]
|
||||||
|
> 本次改动涉及实时通信核心路径,建议在 staging 环境部署后观察至少 30 分钟:
|
||||||
|
> 1. 检查节点在面板中状态是否正常显示为在线
|
||||||
|
> 2. 手动停止面板后观察 Agent 日志,确认重连间隔呈指数增长
|
||||||
|
> 3. 恢复面板后确认 Agent 能自动恢复连接并恢复指标上报
|
||||||
|
> 4. 通过面板下发命令(如添加/删除 service),确认命令执行成功
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 任务清单
|
||||||
|
|
||||||
|
- [x] 优化 1:Agent 增加独立 WebSocket ping 发送
|
||||||
|
- [x] 优化 2:Agent 指数退避重连
|
||||||
|
- [x] 优化 3:Panel 缓存 AES 加密器
|
||||||
|
- [x] 优化 4:Agent 减少 receiveMessages 锁竞争
|
||||||
|
- [x] 优化 5:Agent 命令处理统一异步化
|
||||||
|
- [x] 运行现有测试验证不回归
|
||||||
Reference in New Issue
Block a user