From bcaa4563acdbd8532809a1f3aa0bd4ac47c096f1 Mon Sep 17 00:00:00 2001
From: "meilin.huang" <954537473@qq.com>
Date: Tue, 27 May 2025 22:56:54 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20ssh=20tunnel=E6=A3=80=E6=B5=8B=E5=AF=BC?=
 =?UTF-8?q?=E8=87=B4=E6=AD=BB=E9=94=81=E9=97=AE=E9=A2=98=E8=B0=83=E6=95=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 frontend/src/views/ops/redis/Info.vue         |  9 ++-
 server/internal/db/dbm/dbm.go                 |  2 +-
 server/internal/es/application/es_instance.go |  2 +-
 server/internal/machine/mcm/client_cache.go   |  9 ++-
 server/internal/mongo/mgm/conn_cache.go       |  2 +-
 server/internal/redis/rdm/conn_cache.go       |  2 +-
 server/pkg/pool/cache_pool.go                 | 61 ++++++++++++++-----
 server/pkg/pool/chan_pool.go                  | 15 +++--
 server/pkg/pool/config.go                     | 17 ++++++
 server/pkg/pool/pool.go                       |  2 +-
 10 files changed, 93 insertions(+), 28 deletions(-)

diff --git a/frontend/src/views/ops/redis/Info.vue b/frontend/src/views/ops/redis/Info.vue
index c5307fb0..b78b39b1 100644
--- a/frontend/src/views/ops/redis/Info.vue
+++ b/frontend/src/views/ops/redis/Info.vue
@@ -1,6 +1,13 @@
 <template>
     <div>
-        <el-dialog :title="title" v-model="dialogVisible" :show-close="true" width="1000px" @close="close()">
+        <el-dialog
+            :title="title"
+            v-model="dialogVisible"
+            :show-close="true"
+            width="1000px"
+            @close="close()"
+            body-class="h-[65vh] overflow-y-auto overflow-x-hidden"
+        >
             <el-row :gutter="20">
                 <el-col :lg="16" :md="16">
                     <el-descriptions class="redis-info info-server" :title="$t('redis.redisInfoTitle')" :column="3" size="small" border>
diff --git a/server/internal/db/dbm/dbm.go b/server/internal/db/dbm/dbm.go
index 270f88f2..7164d0ae 100644
--- a/server/internal/db/dbm/dbm.go
+++ b/server/internal/db/dbm/dbm.go
@@ -25,7 +25,7 @@ func init() {
 			if v.Stats().TotalConns == 0 {
 				continue // 连接池中没有连接，跳过
 			}
-			conn, err := v.Get(context.Background())
+			conn, err := v.Get(context.Background(), pool.WithNoUpdateLastActive())
 			if err != nil {
 				continue // 获取连接失败，跳过
 			}
diff --git a/server/internal/es/application/es_instance.go b/server/internal/es/application/es_instance.go
index 60e478ee..7998c990 100644
--- a/server/internal/es/application/es_instance.go
+++ b/server/internal/es/application/es_instance.go
@@ -48,7 +48,7 @@ func init() {
 			if v.Stats().TotalConns == 0 {
 				continue // 连接池中没有连接，跳过
 			}
-			conn, err := v.Get(context.Background())
+			conn, err := v.Get(context.Background(), pool.WithNoUpdateLastActive())
 			if err != nil {
 				continue // 获取连接失败，跳过
 			}
diff --git a/server/internal/machine/mcm/client_cache.go b/server/internal/machine/mcm/client_cache.go
index 57f848c2..f49bc201 100644
--- a/server/internal/machine/mcm/client_cache.go
+++ b/server/internal/machine/mcm/client_cache.go
@@ -2,6 +2,7 @@ package mcm
 
 import (
 	"context"
+	"fmt"
 	"mayfly-go/pkg/pool"
 )
 
@@ -51,11 +52,11 @@ func GetMachineCli(ctx context.Context, authCertName string, getMachine func(str
 
 // 删除指定机器缓存客户端，并关闭客户端连接
 func DeleteCli(id uint64) {
-	for _, pool := range poolGroup.AllPool() {
-		if pool.Stats().TotalConns == 0 {
+	for _, p := range poolGroup.AllPool() {
+		if p.Stats().TotalConns == 0 {
 			continue
 		}
-		conn, err := pool.Get(context.Background())
+		conn, err := p.Get(context.Background(), pool.WithNoUpdateLastActive())
 		if err != nil {
 			continue
 		}
@@ -63,4 +64,6 @@ func DeleteCli(id uint64) {
 			poolGroup.Close(conn.Info.AuthCertName)
 		}
 	}
+	// 删除隧道
+	tunnelPoolGroup.Close(fmt.Sprintf("machine-tunnel-%d", id))
 }
diff --git a/server/internal/mongo/mgm/conn_cache.go b/server/internal/mongo/mgm/conn_cache.go
index 3a3cbcba..aecf9cb5 100644
--- a/server/internal/mongo/mgm/conn_cache.go
+++ b/server/internal/mongo/mgm/conn_cache.go
@@ -17,7 +17,7 @@ func init() {
 			if v.Stats().TotalConns == 0 {
 				continue // 连接池中没有连接，跳过
 			}
-			conn, err := v.Get(context.Background())
+			conn, err := v.Get(context.Background(), pool.WithNoUpdateLastActive())
 			if err != nil {
 				continue // 获取连接失败，跳过
 			}
diff --git a/server/internal/redis/rdm/conn_cache.go b/server/internal/redis/rdm/conn_cache.go
index 4ffe1860..fe3dd702 100644
--- a/server/internal/redis/rdm/conn_cache.go
+++ b/server/internal/redis/rdm/conn_cache.go
@@ -18,7 +18,7 @@ func init() {
 			if v.Stats().TotalConns == 0 {
 				continue // 连接池中没有连接，跳过
 			}
-			rc, err := v.Get(context.Background())
+			rc, err := v.Get(context.Background(), pool.WithNoUpdateLastActive())
 			if err != nil {
 				continue // 获取连接失败，跳过
 			}
diff --git a/server/pkg/pool/cache_pool.go b/server/pkg/pool/cache_pool.go
index a02aa651..f16126ef 100644
--- a/server/pkg/pool/cache_pool.go
+++ b/server/pkg/pool/cache_pool.go
@@ -2,6 +2,7 @@ package pool
 
 import (
 	"context"
+	"math/rand"
 	"mayfly-go/pkg/logx"
 	"mayfly-go/pkg/utils/stringx"
 	"sync"
@@ -45,20 +46,36 @@ func NewCachePool[T Conn](factory func() (T, error), opts ...Option[T]) *CachePo
 }
 
 // Get 获取连接（自动创建或复用缓存连接）
-func (p *CachePool[T]) Get(ctx context.Context) (T, error) {
+func (p *CachePool[T]) Get(ctx context.Context, opts ...GetOption) (T, error) {
 	var zero T
 
+	options := getOptions{updateLastActive: true} // 默认更新 lastActive
+	for _, apply := range opts {
+		apply(&options)
+	}
+
 	// 先尝试加读锁，仅用于查找可用连接
 	p.mu.RLock()
-	for _, entry := range p.cache {
-		if time.Since(entry.lastActive) <= p.config.IdleTimeout {
-			p.mu.RUnlock() // 找到后释放读锁
-			return entry.conn, nil
+	if len(p.cache) >= p.config.MaxConns {
+		keys := make([]string, 0, len(p.cache))
+		for k := range p.cache {
+			keys = append(keys, k)
 		}
+
+		randomKey := keys[rand.Intn(len(keys))]
+		entry := p.cache[randomKey]
+		conn := entry.conn
+
+		if options.updateLastActive {
+			// 更新最后活跃时间
+			entry.lastActive = time.Now()
+		}
+		p.mu.RUnlock()
+		return conn, nil
 	}
 	p.mu.RUnlock()
 
-	// 没有找到可用连接，升级为写锁进行清理和创建
+	// 没有找到可用连接，升级为写锁进行创建
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -66,14 +83,13 @@ func (p *CachePool[T]) Get(ctx context.Context) (T, error) {
 		return zero, ErrPoolClosed
 	}
 
-	// 再次检查是否已有可用连接（防止并发创建）
-	for key, entry := range p.cache {
-		if time.Since(entry.lastActive) <= p.config.IdleTimeout {
-			entry.lastActive = time.Now()
-			return entry.conn, nil
-		}
-		// 清理超时连接
-		if !p.closeConn(key, entry, false) {
+	// 再次检查是否已创建（防止并发）
+	if len(p.cache) >= p.config.MaxConns {
+		for _, entry := range p.cache {
+			if options.updateLastActive {
+				// 更新最后活跃时间
+				entry.lastActive = time.Now()
+			}
 			return entry.conn, nil
 		}
 	}
@@ -206,7 +222,7 @@ func (p *CachePool[T]) cleanupIdle() {
 
 	cutoff := time.Now().Add(-p.config.IdleTimeout)
 	for key, entry := range p.cache {
-		if entry.lastActive.Before(cutoff) || entry.conn.Ping() != nil {
+		if entry.lastActive.Before(cutoff) || !p.ping(entry.conn) {
 			logx.Infof("cache pool - cleaning up idle connection, key: %s", key)
 			// 如果连接超时或不可用，则关闭连接
 			p.closeConn(key, entry, false)
@@ -214,6 +230,21 @@ func (p *CachePool[T]) cleanupIdle() {
 	}
 }
 
+func (p *CachePool[T]) ping(conn T) bool {
+	done := make(chan struct{})
+	var result bool
+	go func() {
+		result = conn.Ping() == nil
+		close(done)
+	}()
+	select {
+	case <-done:
+		return result
+	case <-time.After(2 * time.Second): // 设置超时
+		return false // 超时认为不可用
+	}
+}
+
 func (p *CachePool[T]) closeConn(key string, entry *cacheEntry[T], force bool) bool {
 	if !force {
 		// 如果不是强制关闭且有连接关闭回调，则调用回调
diff --git a/server/pkg/pool/chan_pool.go b/server/pkg/pool/chan_pool.go
index ae64e939..46b1f37c 100644
--- a/server/pkg/pool/chan_pool.go
+++ b/server/pkg/pool/chan_pool.go
@@ -74,12 +74,17 @@ func NewChannelPool[T Conn](factory func() (T, error), opts ...Option[T]) *ChanP
 	return p
 }
 
-func (p *ChanPool[T]) Get(ctx context.Context) (T, error) {
+func (p *ChanPool[T]) Get(ctx context.Context, opts ...GetOption) (T, error) {
 	connChan := make(chan T, 1)
 	errChan := make(chan error, 1)
 
+	options := getOptions{updateLastActive: true} // 默认更新 lastActive
+	for _, apply := range opts {
+		apply(&options)
+	}
+
 	go func() {
-		conn, err := p.get()
+		conn, err := p.get(options)
 		if err != nil {
 			errChan <- err
 		} else {
@@ -108,7 +113,7 @@ func (p *ChanPool[T]) Get(ctx context.Context) (T, error) {
 	}
 }
 
-func (p *ChanPool[T]) get() (T, error) {
+func (p *ChanPool[T]) get(opts getOptions) (T, error) {
 	// 检查连接池是否已关闭
 	p.mu.RLock()
 	if p.closed {
@@ -123,7 +128,9 @@ func (p *ChanPool[T]) get() (T, error) {
 	case wrapper := <-p.idleConns:
 		atomic.AddInt32(&p.stats.IdleConns, -1)
 		atomic.AddInt32(&p.stats.ActiveConns, 1)
-		wrapper.lastActive = time.Now()
+		if opts.updateLastActive {
+			wrapper.lastActive = time.Now()
+		}
 		return wrapper.conn, nil
 	default:
 		return p.createConn()
diff --git a/server/pkg/pool/config.go b/server/pkg/pool/config.go
index 12c76b61..82714e78 100644
--- a/server/pkg/pool/config.go
+++ b/server/pkg/pool/config.go
@@ -62,3 +62,20 @@ func WithOnConnClose[T Conn](fn func(conn T) error) Option[T] {
 		c.OnConnClose = fn
 	}
 }
+
+/**** GetOption Config ****/
+
+// GetOption 用于配置 Get 的行为
+type GetOption func(*getOptions)
+
+// 控制 Get 行为的选项
+type getOptions struct {
+	updateLastActive bool // 是否更新 lastActive，默认 true
+}
+
+// WithNoUpdateLastActive 返回一个 Option，禁用更新 lastActive
+func WithNoUpdateLastActive() GetOption {
+	return func(o *getOptions) {
+		o.updateLastActive = false
+	}
+}
diff --git a/server/pkg/pool/pool.go b/server/pkg/pool/pool.go
index cab6ea03..8d3126ee 100644
--- a/server/pkg/pool/pool.go
+++ b/server/pkg/pool/pool.go
@@ -17,7 +17,7 @@ type Conn interface {
 // Pool 连接池接口
 type Pool[T Conn] interface {
 	// 核心方法
-	Get(ctx context.Context) (T, error)
+	Get(ctx context.Context, opts ...GetOption) (T, error)
 	Put(T) error
 	Close()