同步完整源码至远程:扁平化重构现状 + 未入库源文件

- 补入库:server/internal/db/dump.go、redis 分页/扫描预算测试、opsxdb-mcp main_test、
  packages/opsxdb-mcp package-lock、docs 审计与对比文档等 37 个新文件
- 含结构扁平化进行中的 80 处修改与 155 处旧结构删除(server/ src/ src-tauri/ 为当前有效源)
- 构建产物(bin/target/dist)与签名私钥仍按规则排除
This commit is contained in:
logan committed 2026-10-16 03:47:33 +08:00
1 parent 5c0be49217
commit 2c0bfafd8d
269 files changed
+10372 -16435

No files matched your search

+44 -26
View File
@@ -18,10 +18,10 @@ import (
"crypto/sha1"
"crypto/sha256"
"encoding/hex"
"hash"
"encoding/json"
"errors"
"fmt"
"hash"
"io"
"net/http"
"os"
@@ -215,13 +215,9 @@ func (b *cappedBuffer) String() string {
// StartAgent 按 spec 启动(或复用)JVM agent 进程并完成握手。
// agent jar 与开源驱动 jar 在首次连接时按需下载(DMG 不打包 Java 产物,保持安装包瘦身)。
func StartAgent(spec AgentSpec) (*AgentClient, error) {
javaBin := os.Getenv("OPSXDB_JAVA_BIN")
if javaBin == "" {
p, err := exec.LookPath("java")
if err != nil {
return nil, errors.New("未找到 java:请安装 JDK 17+ 或设置 OPSXDB_JAVA_BIN")
}
javaBin = p
javaBin, _, err := agentJavaRuntime()
if err != nil {
return nil, err
}
if err := ensureAgentRuntime(spec); err != nil {
return nil, err
@@ -1019,27 +1015,21 @@ type JarStatus struct {
type DriverStatus struct {
ID string `json:"id"`
Label string `json:"label"`
Runtime string `json:"runtime"` // agent
License string `json:"license"` // oss / proprietary
AutoFetch bool `json:"autoFetch"` // 开源可自动下载
Runtime string `json:"runtime"` // agent
License string `json:"license"` // oss / proprietary
AutoFetch bool `json:"autoFetch"` // 开源可自动下载
Jars []JarStatus `json:"jars"`
AllInstalled bool `json:"allInstalled"`
JDKRequired bool `json:"jdkRequired"` // agent 驱动恒 true
JDKRequired bool `json:"jdkRequired"` // agent 驱动恒 true
JDKOK bool `json:"jdkOk"`
JavaVersion string `json:"javaVersion"` // 探测到的版本(如 21.0.12)
AgentInstalled bool `json:"agentInstalled"` // agent jar 是否就位
OfflinePack bool `json:"offlinePack"` // 离线驱动包是否就位
AgentInstalled bool `json:"agentInstalled"` // agent jar 是否就位
OfflinePack bool `json:"offlinePack"` // 离线驱动包是否就位
}
// DriversStatus 汇总所有 agent 驱动的安装状态,供驱动管理界面展示。
func DriversStatus() []DriverStatus {
javaBin, _ := exec.LookPath("java")
javaVersion := ""
if javaBin != "" {
if out, err := exec.Command(javaBin, "-version").CombinedOutput(); err == nil {
javaVersion = parseJavaVersion(string(out))
}
}
_, javaVersion, javaErr := agentJavaRuntime()
_, offlineErr := os.Stat(offlinePackPath())
out := []DriverStatus{}
@@ -1057,7 +1047,7 @@ func DriversStatus() []DriverStatus {
Runtime: "agent",
AutoFetch: auto,
JDKRequired: true,
JDKOK: javaBin != "",
JDKOK: javaErr == nil,
JavaVersion: javaVersion,
OfflinePack: offlineErr == nil,
AgentInstalled: fileExists(filepath.Join(agentDir(), agentJarName)),
@@ -1103,10 +1093,8 @@ func InstallDriver(driverKey string) error {
if !ok {
return fmt.Errorf("未知 agent 驱动: %s", driverKey)
}
if _, err := exec.LookPath(os.Getenv("OPSXDB_JAVA_BIN")); err != nil && os.Getenv("OPSXDB_JAVA_BIN") == "" {
if _, err := exec.LookPath("java"); err != nil {
return fmt.Errorf("未找到 java,请先安装 JDK 17+")
}
if _, _, err := agentJavaRuntime(); err != nil {
return err
}
return ensureAgentRuntime(spec)
}
@@ -1159,6 +1147,9 @@ func ImportOfflinePack(zipPath string) (int, error) {
if !strings.HasPrefix(rel, "agent/") && !strings.HasPrefix(rel, "libs/") && !strings.HasPrefix(rel, "drivers/") {
continue
}
if !filepath.IsLocal(filepath.FromSlash(rel)) {
return imported, fmt.Errorf("离线包包含非法路径: %s", rel)
}
rc, err := f.Open()
if err != nil {
return imported, err
@@ -1209,3 +1200,30 @@ func parseJavaVersion(out string) string {
}
return ""
}
// agentJavaRuntime 与实际启动使用同一 Java 路径,避免设置页把旧版 Java 误报为就绪。
func agentJavaRuntime() (string, string, error) {
javaBin := os.Getenv("OPSXDB_JAVA_BIN")
if javaBin == "" {
javaBin = "java"
}
path, err := exec.LookPath(javaBin)
if err != nil {
return "", "", errors.New("未找到 Java,请安装 Java 17+ 或设置 OPSXDB_JAVA_BIN")
}
out, err := exec.Command(path, "-version").CombinedOutput()
version := parseJavaVersion(string(out))
if err != nil {
return path, version, fmt.Errorf("Java 运行失败: %w", err)
}
major := 0
if strings.HasPrefix(version, "1.") {
fmt.Sscanf(version, "1.%d", &major)
} else {
fmt.Sscanf(version, "%d", &major)
}
if major < 17 {
return path, version, fmt.Errorf("Java %s 不受支持,agent 需要 Java 17+", version)
}
return path, version, nil
}
+40
View File
@@ -102,6 +102,46 @@ func runAgentStub() {
// 不会走到这里(TestMain 已拦截),但名字必须存在以匹配 -test.run。
func TestAgentStubHelper(t *testing.T) {}
func TestAgentJavaRuntimeVersion(t *testing.T) {
path := filepath.Join(t.TempDir(), "java")
t.Setenv("OPSXDB_JAVA_BIN", path)
for _, tc := range []struct {
version string
wantOK bool
}{
{"1.8.0_401", false},
{"17.0.12", true},
{"21.0.2", true},
} {
script := "#!/bin/sh\necho 'openjdk version \"" + tc.version + "\"' >&2\n"
if err := os.WriteFile(path, []byte(script), 0o755); err != nil {
t.Fatal(err)
}
_, gotVersion, err := agentJavaRuntime()
if (err == nil) != tc.wantOK || gotVersion != tc.version {
t.Fatalf("Java %s: version=%q err=%v", tc.version, gotVersion, err)
}
}
}
func TestImportOfflinePackRejectsTraversal(t *testing.T) {
home := t.TempDir()
t.Setenv("OPSXDB_HOME", home)
content := "unsafe"
rel := "drivers/../../escape.jar"
manifest := []map[string]string{{"path": rel, "sha256": fmt.Sprintf("%x", sha256.Sum256([]byte(content)))}}
path := filepath.Join(home, "malicious.zip")
if err := writeTestZip(path, manifest, map[string]string{rel: content}); err != nil {
t.Fatal(err)
}
if _, err := ImportOfflinePack(path); err == nil {
t.Fatal("离线驱动包不得写到驱动目录之外")
}
if _, err := os.Stat(filepath.Join(home, "escape.jar")); !os.IsNotExist(err) {
t.Fatalf("导入越界文件: %v", err)
}
}
func startStubAgent(t *testing.T) *AgentClient {
t.Helper()
cmd := exec.Command(os.Args[0], "-test.run=TestAgentStubHelper", "--", "-agent-stub")
+77
View File
@@ -0,0 +1,77 @@
package db
import (
"context"
"fmt"
"io"
"strings"
"time"
)
// DumpMode 表转储范围:仅结构 / 仅数据 / 结构+数据(对齐 Navicat「转储 SQL 文件」三态)。
const (
DumpStructure = "structure"
DumpData = "data"
DumpBoth = "structure_data"
)
// DumpTable 把单表转储为 .sql 写入 w(流式,数据部分复用 ExportQuery 的 insert 编码)。
// 不自动附带 DROP TABLE,避免导入时误删既有表;注释头标注来源与时间便于追溯。
func DumpTable(ctx context.Context, c ConnReq, w io.Writer) error {
mode := strings.ToLower(strings.TrimSpace(c.Format))
if mode == "" {
mode = DumpBoth
}
if mode != DumpStructure && mode != DumpData && mode != DumpBoth {
return fmt.Errorf("mode 仅支持 structure/data/structure_data")
}
if strings.TrimSpace(c.Table) == "" {
return fmt.Errorf("缺少表名")
}
// table 支持 "库.表" 形式:拆出 database,避免整体被引号包住导致 No database selected
if i := strings.Index(c.Table, "."); i > 0 {
if strings.TrimSpace(c.Database) == "" {
c.Database = c.Table[:i]
}
c.Table = c.Table[i+1:]
}
// 数据部分复用 insert 编码:值拷贝整个 ConnReq,只覆盖导出相关字段
dataReq := c
dataReq.Format = "insert"
dataReq.SQL = SelectAllSQL(c.Driver, c.Table)
if mode == DumpData {
return ExportQuery(ctx, dataReq, w)
}
var b strings.Builder
fmt.Fprintf(&b, "-- OpsxDB 表转储 %s\n-- 表: %s\n-- 时间: %s\n\n",
modeLabel(mode), c.Table, time.Now().Format("2006-01-02 15:04:05"))
ddl, err := GetDDL(ctx, c)
if err != nil {
return err
}
b.WriteString(strings.TrimRight(ddl, ";\n"))
b.WriteString(";\n")
if mode == DumpBoth {
b.WriteString("\n")
}
if _, err := io.WriteString(w, b.String()); err != nil {
return err
}
if mode == DumpBoth {
if err := ExportQuery(ctx, dataReq, w); err != nil {
return err
}
}
return nil
}
func modeLabel(mode string) string {
switch mode {
case DumpStructure:
return "· 仅结构"
case DumpData:
return "· 仅数据"
default:
return "· 结构和数据"
}
}
+1 -1
View File
@@ -152,7 +152,7 @@ func listColumnsMySQL(ctx context.Context, db *sql.DB, database, table string) (
if err := rows.Scan(&field, &typ, &collation, &null, &key, &def, &extra, &priv, &comment); err != nil {
return nil, err
}
col := ColumnInfo{Name: field, Type: typ, Nullable: null == "YES", Key: key, Comment: comment.String}
col := ColumnInfo{Name: field, Type: typ, Nullable: null == "YES", Key: key, Comment: comment.String, Extra: extra.String}
if def.Valid {
col.Default = def.String
}
+239 -11
View File
@@ -2,10 +2,15 @@ package db
import (
"context"
"crypto/rand"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/redis/go-redis/v9"
@@ -24,6 +29,11 @@ func redisDBIndex(c ConnReq) int {
func normalizeRedisMode(c ConnReq) string {
m := strings.ToLower(strings.TrimSpace(c.RedisMode))
if m == "" {
// 未显式选模式但填了多个节点:按 Cluster 处理(Sentinel 必须显式选并填 masterName,
// 不会误伤)。避免"填了种子节点但忘切模式"时按单机连首节点、单键操作 MOVED 直抛。
if len(parseRedisAddrs(c.RedisNodes)) > 1 {
return "cluster"
}
return "standalone"
}
return m
@@ -183,7 +193,7 @@ func testRedis(c ConnReq) error {
defer cli.Close()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
return cli.Ping(ctx).Err()
return redisConnectionError(c, cli.Ping(ctx).Err())
}
func listDatabasesRedis(parent context.Context, c ConnReq) ([]string, error) {
@@ -196,7 +206,7 @@ func listDatabasesRedis(parent context.Context, c ConnReq) ([]string, error) {
ctx, cancel := reqCtxRedis(parent, 5*time.Second)
defer cancel()
if err := cli.Ping(ctx).Err(); err != nil {
return nil, err
return nil, redisConnectionError(c, err)
}
// Cluster 不支持 SELECT 多库,只暴露 0
if normalizeRedisMode(c) == "cluster" {
@@ -264,7 +274,12 @@ func redisAffected(start time.Time, n int64) ExecResult {
}
func redisErr(start time.Time, err error) ExecResult {
return ExecResult{Error: errMsg(err), TimeMs: time.Since(start).Milliseconds()}
msg := errMsg(err)
// 单机模式客户端不做 MOVED 重定向:给出可操作提示,引导切换部署模式
if strings.Contains(msg, "MOVED ") || strings.Contains(msg, "ASK ") {
msg += ";该键属于集群其他分片,当前按单机模式连接。请在连接表单把部署模式切为 Cluster 并填写种子节点。"
}
return ExecResult{Error: msg, TimeMs: time.Since(start).Milliseconds()}
}
func redisUsage(start time.Time, usage string) ExecResult {
@@ -322,7 +337,7 @@ func execRedis(parent context.Context, c ConnReq) ExecResult {
return ExecResult{Columns: []string{"key", "value"}, Rows: []map[string]interface{}{{"key": args[0], "value": val}}, RowCount: 1, TimeMs: time.Since(start).Milliseconds()}
case "SET":
// 参数已按 shell 词法切分,value 即 args[1](不再 Join 破坏引号分组);
// 支持可选 EX 秒数:SET key value [EX n]
// 支持 EX 或 KEEPTTL;详情页保存需保持原键的过期时间。
if len(args) < 2 {
return redisUsage(start, "SET key value [EX seconds]")
}
@@ -334,10 +349,54 @@ func execRedis(parent context.Context, c ConnReq) ExecResult {
}
exp = time.Duration(n) * time.Second
}
if len(args) == 3 && strings.EqualFold(args[2], "KEEPTTL") {
if err := cli.SetArgs(ctx, args[0], args[1], redis.SetArgs{KeepTTL: true}).Err(); err != nil {
if !strings.Contains(strings.ToLower(err.Error()), "syntax error") {
return redisErr(start, err)
}
// Redis <6 无 KEEPTTL:原子读取 PTTL 再写回,避免分开的命令丢失过期时间。
const keepTTLScript = `local kind = redis.call('TYPE', KEYS[1])
if kind.ok ~= 'string' then return redis.error_reply('键已变化,请刷新后重试') end
local ttl = redis.call('PTTL', KEYS[1])
redis.call('SET', KEYS[1], ARGV[1])
if ttl > 0 then redis.call('PEXPIRE', KEYS[1], ttl) end
return 'OK'`
if scriptErr := redis.NewScript(keepTTLScript).Run(ctx, cli, []string{args[0]}, args[1]).Err(); scriptErr != nil {
return redisErr(start, scriptErr)
}
}
return redisOK(start)
}
if len(args) > 2 && !(len(args) == 4 && strings.EqualFold(args[2], "EX")) {
return redisUsage(start, "SET key value [EX seconds | KEEPTTL]")
}
if err := cli.Set(ctx, args[0], args[1], exp).Err(); err != nil {
return redisErr(start, err)
}
return redisOK(start)
case "LDELETEINDEX":
// 内部单行操作:按索引和当前值校验后,在同一脚本中移除该位置。
// LREM 1 value 会误删前面同值元素;随机标记避免与现有列表值冲突。
if len(args) != 3 {
return redisUsage(start, "LDELETEINDEX key index expectedValue")
}
index, parseErr := strconv.ParseInt(args[1], 10, 64)
if parseErr != nil || index < 0 {
return redisUsage(start, "LDELETEINDEX index 需为非负整数")
}
random := make([]byte, 24)
if _, err := rand.Read(random); err != nil {
return redisErr(start, err)
}
marker := "\x00opsxdb-delete:" + hex.EncodeToString(random)
const script = `local value = redis.call('LINDEX', KEYS[1], ARGV[1])
if value == false or value ~= ARGV[2] then return redis.error_reply('列表已变化,请刷新后重试') end
redis.call('LSET', KEYS[1], ARGV[1], ARGV[3])
return redis.call('LREM', KEYS[1], 1, ARGV[3])`
if err := redis.NewScript(script).Run(ctx, cli, []string{args[0]}, index, args[2], marker).Err(); err != nil {
return redisErr(start, err)
}
return redisOK(start)
case "DEL":
if len(args) < 1 {
return redisUsage(start, "DEL key [key...]")
@@ -825,6 +884,17 @@ func redisScanPattern(c ConnReq) string {
return p
}
func redisPatHasGlob(p string) bool {
return strings.ContainsAny(p, "*?[")
}
func redisBrowsePattern(p string) string {
if p == "*" || redisPatHasGlob(p) {
return p
}
return "*" + p + "*"
}
func redisValueLimit(c ConnReq) int {
n := c.Limit
if n <= 0 {
@@ -850,10 +920,23 @@ func ttlSeconds(d time.Duration) int64 {
// scanAll 全量 SCAN 到 limit 截断。集群模式遍历所有 master 分片 ——
// ClusterClient.Scan 只命中单个节点,会漏掉其他分片的键。
func scanAll(ctx context.Context, cli redis.UniversalClient, pat string, count int64, limit int) (names []string, truncated bool, err error) {
func scanAll(ctx context.Context, cli redis.UniversalClient, pat string, count int64, limit int, scanBudget ...int) (names []string, truncated bool, err error) {
var mu sync.Mutex
calls := 0
scanOne := func(c scanClient) error {
mu.Lock()
defer mu.Unlock()
if len(names) >= limit {
truncated = true
return nil
}
var cursor uint64
for {
if len(scanBudget) > 0 && calls >= scanBudget[0] {
truncated = true
return nil
}
calls++
batch, next, err := c.Scan(ctx, cursor, pat, count).Result()
if err != nil {
return err
@@ -883,6 +966,9 @@ func scanAll(ctx context.Context, cli redis.UniversalClient, pat string, count i
return names, truncated, err
}
err = scanOne(cli)
if len(names) > limit {
names = names[:limit]
}
return names, truncated, err
}
@@ -891,7 +977,62 @@ type scanClient interface {
Scan(ctx context.Context, cursor uint64, match string, count int64) *redis.ScanCmd
}
type keyScanPosition struct {
Pattern string `json:"p"`
DB string `json:"d"`
Nodes []string `json:"n"`
Index int `json:"i"`
Cursor uint64 `json:"c"`
}
// Each page carries its shard and SCAN cursor. A complete SCAN batch is kept
// together so a server-side COUNT hint returning >limit keys cannot drop keys.
func scanKeyPage(ctx context.Context, clients []scanClient, pos keyScanPosition) ([]string, keyScanPosition, error) {
names := make([]string, 0, redisKeyListCap)
calls := 0
for pos.Index < len(clients) && calls < 20 && len(names) < redisKeyListCap {
batch, next, err := clients[pos.Index].Scan(ctx, pos.Cursor, pos.Pattern, 200).Result()
if err != nil {
return nil, pos, err
}
calls++
names = append(names, batch...)
if next == 0 {
pos.Index++
pos.Cursor = 0
} else {
pos.Cursor = next
}
}
return names, pos, nil
}
func keyScanClients(ctx context.Context, cli redis.UniversalClient) ([]scanClient, []string, error) {
if cc, ok := cli.(*redis.ClusterClient); ok {
masters := map[string]*redis.Client{}
err := cc.ForEachMaster(ctx, func(_ context.Context, mc *redis.Client) error {
masters[mc.Options().Addr] = mc
return nil
})
if err != nil {
return nil, nil, err
}
addresses := make([]string, 0, len(masters))
for addr := range masters {
addresses = append(addresses, addr)
}
sort.Strings(addresses)
clients := make([]scanClient, 0, len(addresses))
for _, addr := range addresses {
clients = append(clients, masters[addr])
}
return clients, addresses, nil
}
return []scanClient{cli}, []string{"single"}, nil
}
// ListRedisKeys SCAN + pipeline TYPE/TTL. Cap ~500.
// 完整键名优先展示,同时按 *关键词* 做有预算的模糊 SCAN;截断不代表完整搜索。
func ListRedisKeys(parent context.Context, c ConnReq) RedisKeysResult {
start := time.Now()
cli, done, err := redisClientFor(c)
@@ -903,17 +1044,67 @@ func ListRedisKeys(parent context.Context, c ConnReq) RedisKeysResult {
defer cancel()
pat := redisScanPattern(c)
names, truncated, err := scanAll(ctx, cli, pat, 200, redisKeyListCap)
scanPat := pat
var exact *RedisKeyInfo
exactHit := false
if pat != "*" && pat != "" && !redisPatHasGlob(pat) {
if typ, err := cli.Type(ctx, pat).Result(); err == nil && typ != "none" {
ttl := int64(-2)
if d, err := cli.TTL(ctx, pat).Result(); err == nil {
ttl = ttlSeconds(d)
}
exact = &RedisKeyInfo{Name: pat, Type: typ, TTL: ttl, Exact: true}
exactHit = true
}
// 无通配符按包含匹配;精确 Key 单独置顶。
scanPat = redisBrowsePattern(pat)
}
clients, nodes, err := keyScanClients(ctx, cli)
if err != nil {
return RedisKeysResult{Keys: []RedisKeyInfo{}, Error: errMsg(err), TimeMs: time.Since(start).Milliseconds()}
}
pos := keyScanPosition{Pattern: scanPat, DB: c.Database, Nodes: nodes}
if c.KeyCursor != "" {
decoded, decodeErr := base64.RawURLEncoding.DecodeString(c.KeyCursor)
if decodeErr != nil || len(decoded) > 4096 || json.Unmarshal(decoded, &pos) != nil ||
pos.Pattern != scanPat || pos.DB != c.Database || !equalScanNodes(pos.Nodes, nodes) ||
pos.Index < 0 || pos.Index >= len(clients) {
return RedisKeysResult{Keys: []RedisKeyInfo{}, Error: "搜索位置已失效,请重新搜索", TimeMs: time.Since(start).Milliseconds()}
}
}
names, pos, err := scanKeyPage(ctx, clients, pos)
if err != nil {
return RedisKeysResult{Keys: []RedisKeyInfo{}, Error: errMsg(err), TimeMs: time.Since(start).Milliseconds()}
}
truncated := pos.Index < len(clients)
nextCursor := ""
if truncated {
encoded, _ := json.Marshal(pos)
nextCursor = base64.RawURLEncoding.EncodeToString(encoded)
}
sort.Strings(names)
if names == nil {
names = []string{}
}
out := make([]RedisKeyInfo, len(names))
// 精确命中去重:SCAN 也可能扫到同一键
if exact != nil && c.KeyCursor == "" {
filtered := names[:0]
for _, n := range names {
if n != exact.Name {
filtered = append(filtered, n)
}
}
names = filtered
}
out := make([]RedisKeyInfo, 0, len(names)+1)
if exact != nil && c.KeyCursor == "" {
out = append(out, *exact)
}
if len(names) == 0 {
return RedisKeysResult{Keys: out, Truncated: truncated, TimeMs: time.Since(start).Milliseconds()}
return RedisKeysResult{Keys: out, ExactHit: exactHit, Truncated: truncated, NextCursor: nextCursor, TimeMs: time.Since(start).Milliseconds()}
}
// Pipeline TYPE + TTL (avoid N+1)
@@ -934,9 +1125,21 @@ func ListRedisKeys(parent context.Context, c ConnReq) RedisKeysResult {
if d, err := ttlCmds[i].Result(); err == nil {
ttl = ttlSeconds(d)
}
out[i] = RedisKeyInfo{Name: k, Type: typ, TTL: ttl}
out = append(out, RedisKeyInfo{Name: k, Type: typ, TTL: ttl})
}
return RedisKeysResult{Keys: out, Truncated: truncated, TimeMs: time.Since(start).Milliseconds()}
return RedisKeysResult{Keys: out, ExactHit: exactHit, Truncated: truncated, NextCursor: nextCursor, TimeMs: time.Since(start).Milliseconds()}
}
func equalScanNodes(a, b []string) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
// GetRedisKey typed value → ResultPane columns/rows.
@@ -1152,13 +1355,29 @@ func DeleteRedisPattern(parent context.Context, c ConnReq) RedisDelPatternResult
if len(names) > redisDelPatternCap {
return fail(fmt.Sprintf("匹配键 %d 个超过上限 %d,请缩小目录范围", len(names), redisDelPatternCap))
}
// 集群下一次 UNLINK 多 key 会 CROSSSLOT;按单 key 删除。单机仍可批量。
_, isCluster := cli.(*redis.ClusterClient)
total := 0
for i := 0; i < len(names); i += 500 {
end := i + 500
if end > len(names) {
end = len(names)
}
n, err := cli.Unlink(ctx, names[i:end]...).Result()
batch := names[i:end]
if isCluster {
for _, k := range batch {
n, err := cli.Unlink(ctx, k).Result()
if err != nil {
return RedisDelPatternResult{
Deleted: total, Scanned: len(names), Truncated: truncated,
Error: errMsg(err), TimeMs: time.Since(start).Milliseconds(),
}
}
total += int(n)
}
continue
}
n, err := cli.Unlink(ctx, batch...).Result()
if err != nil {
return RedisDelPatternResult{
Deleted: total, Scanned: len(names), Truncated: truncated,
@@ -1173,3 +1392,12 @@ func DeleteRedisPattern(parent context.Context, c ConnReq) RedisDelPatternResult
}
}
func redisConnectionError(c ConnReq, err error) error {
if err == nil {
return nil
}
if normalizeRedisMode(c) == "cluster" && strings.Contains(err.Error(), "connection refused") {
return fmt.Errorf("Redis 集群节点拒绝连接:%w;请确认节点已启动,种子地址及 cluster-announce-ip/port 均能从当前电脑访问;远程集群返回 127.0.0.1 时需修正节点公布地址", err)
}
return err
}
+83
View File
@@ -0,0 +1,83 @@
package db
import (
"context"
"testing"
"github.com/redis/go-redis/v9"
)
type pageScanner struct {
steps map[uint64]struct {
keys []string
next uint64
}
visited []uint64
}
func (s *pageScanner) Scan(ctx context.Context, cursor uint64, _ string, _ int64) *redis.ScanCmd {
s.visited = append(s.visited, cursor)
step := s.steps[cursor]
cmd := redis.NewScanCmd(ctx, nil, "scan")
cmd.SetVal(step.keys, step.next)
return cmd
}
func TestKeyPageResumesAcrossMastersWithoutDroppingBatch(t *testing.T) {
first := &pageScanner{steps: map[uint64]struct {
keys []string
next uint64
}{
0: {[]string{"a", "b"}, 17},
17: {[]string{"c"}, 0},
}}
second := &pageScanner{steps: map[uint64]struct {
keys []string
next uint64
}{
0: {[]string{"d"}, 0},
}}
clients := []scanClient{first, second}
// Simulate a page already at the next cursor of the first shard.
keys, pos, err := scanKeyPage(context.Background(), clients, keyScanPosition{Pattern: "*", Index: 0, Cursor: 17})
if err != nil || len(keys) != 2 || keys[0] != "c" || keys[1] != "d" || pos.Index != 2 || pos.Cursor != 0 {
t.Fatalf("keys=%v pos=%+v err=%v", keys, pos, err)
}
if len(first.visited) != 1 || first.visited[0] != 17 || len(second.visited) != 1 || second.visited[0] != 0 {
t.Fatalf("unexpected scan positions first=%v second=%v", first.visited, second.visited)
}
}
func TestKeyPageStopsAtBudgetWithResumableCursor(t *testing.T) {
s := &pageScanner{steps: map[uint64]struct {
keys []string
next uint64
}{
0: {nil, 1}, 1: {nil, 1},
}}
keys, pos, err := scanKeyPage(context.Background(), []scanClient{s}, keyScanPosition{Pattern: "rare*"})
if err != nil || len(keys) != 0 || len(s.visited) != 20 || pos.Index != 0 || pos.Cursor != 1 {
t.Fatalf("keys=%v calls=%d pos=%+v err=%v", keys, len(s.visited), pos, err)
}
}
func TestKeyPageKeepsOversizedScanBatch(t *testing.T) {
batch := make([]string, redisKeyListCap+3)
for i := range batch {
batch[i] = "key"
}
s := &pageScanner{steps: map[uint64]struct {
keys []string
next uint64
}{
0: {batch, 9}, 9: {[]string{"last"}, 0},
}}
keys, pos, err := scanKeyPage(context.Background(), []scanClient{s}, keyScanPosition{Pattern: "*"})
if err != nil || len(keys) != len(batch) || pos.Cursor != 9 || pos.Index != 0 {
t.Fatalf("first page keys=%d pos=%+v err=%v", len(keys), pos, err)
}
nextKeys, end, err := scanKeyPage(context.Background(), []scanClient{s}, pos)
if err != nil || len(nextKeys) != 1 || nextKeys[0] != "last" || end.Index != 1 {
t.Fatalf("second page keys=%v pos=%+v err=%v", nextKeys, end, err)
}
}
+31
View File
@@ -1,8 +1,10 @@
package db
import (
"errors"
"strings"
"testing"
"time"
)
func TestParseRedisAddrs(t *testing.T) {
@@ -26,3 +28,32 @@ func TestOpenRedisModeValidation(t *testing.T) {
t.Fatalf("cluster without nodes: %v", err)
}
}
func TestNormalizeRedisModeInfersClusterFromNodes(t *testing.T) {
cases := []struct {
name string
c ConnReq
want string
}{
{"空配置回单机", ConnReq{}, "standalone"},
{"显式 sentinel 保留", ConnReq{RedisMode: "Sentinel", RedisNodes: "127.0.0.1:26379"}, "sentinel"},
{"多节点未选模式推断 cluster", ConnReq{RedisNodes: "127.0.0.1:7101,127.0.0.1:7102"}, "cluster"},
{"单节点未选模式保持单机", ConnReq{Host: "127.0.0.1", Port: 6379, RedisNodes: "127.0.0.1:6379"}, "standalone"},
}
for _, tc := range cases {
if got := normalizeRedisMode(tc.c); got != tc.want {
t.Fatalf("%s: want %s, got %s", tc.name, tc.want, got)
}
}
}
func TestRedisErrMovedHint(t *testing.T) {
res := redisErr(time.Now(), errors.New("MOVED 12503 127.0.0.1:7103"))
if !strings.Contains(res.Error, "部署模式切为 Cluster") {
t.Fatalf("MOVED 错误应带模式切换提示: %q", res.Error)
}
res = redisErr(time.Now(), errors.New("WRONGTYPE Operation against"))
if strings.Contains(res.Error, "Cluster") {
t.Fatalf("普通错误不应带集群提示: %q", res.Error)
}
}
@@ -0,0 +1,42 @@
package db
import (
"context"
"github.com/redis/go-redis/v9"
"testing"
)
type budgetRedis struct {
redis.UniversalClient
calls int
batch []string
}
func (c *budgetRedis) Scan(ctx context.Context, cursor uint64, match string, count int64) *redis.ScanCmd {
c.calls++
cmd := redis.NewScanCmd(ctx, nil, "scan")
cmd.SetVal(c.batch, uint64(c.calls))
return cmd
}
func TestRedisScanBudgetSparseMatches(t *testing.T) {
c := &budgetRedis{}
names, truncated, err := scanAll(context.Background(), c, "missing*", 200, 500, 20)
if err != nil || !truncated || len(names) != 0 || c.calls != 20 {
t.Fatalf("calls=%d names=%v truncated=%v err=%v", c.calls, names, truncated, err)
}
}
func TestRedisScanCountIsNotHardLimit(t *testing.T) {
c := &budgetRedis{batch: []string{"a", "b", "c"}}
names, truncated, err := scanAll(context.Background(), c, "*", 1, 2, 20)
if err != nil || !truncated || len(names) != 2 || c.calls != 1 {
t.Fatalf("names=%v calls=%d truncated=%v err=%v", names, c.calls, truncated, err)
}
}
func TestRedisBrowsePattern(t *testing.T) {
for input, want := range map[string]string{"order": "*order*", "order:*": "order:*", "*": "*"} {
if got := redisBrowsePattern(input); got != want {
t.Fatalf("input=%q got=%q want=%q", input, got, want)
}
}
}
+42 -37
View File
@@ -12,19 +12,19 @@ const MaxRowLimit = 10000
// ConnReq 连接 / 执行请求。sessionId 有值时服务端用会话里的账密,请求里的 password 可空。
type ConnReq struct {
Driver string `json:"driver"`
Host string `json:"host"`
Port int `json:"port"`
User string `json:"user"`
Password string `json:"password"`
Database string `json:"database"`
SQL string `json:"sql"`
Service string `json:"service"`
SSL bool `json:"ssl"`
SSLMode string `json:"sslMode"` // disable|require|skip-verify|verify-ca
SSLCA string `json:"sslCA"`
SSLCert string `json:"sslCert"`
SSLKey string `json:"sslKey"`
Driver string `json:"driver"`
Host string `json:"host"`
Port int `json:"port"`
User string `json:"user"`
Password string `json:"password"`
Database string `json:"database"`
SQL string `json:"sql"`
Service string `json:"service"`
SSL bool `json:"ssl"`
SSLMode string `json:"sslMode"` // disable|require|skip-verify|verify-ca
SSLCA string `json:"sslCA"`
SSLCert string `json:"sslCert"`
SSLKey string `json:"sslKey"`
// 证书 PEM 内容,优先于上面的文件路径(客户端与服务端分离部署时免共享文件)
SSLCAContent string `json:"sslCAContent"`
SSLCertContent string `json:"sslCertContent"`
@@ -36,12 +36,13 @@ type ConnReq struct {
SessionID string `json:"sessionId"`
Table string `json:"table"`
Limit int `json:"limit"`
Offset int `json:"offset"` // 结果分页:跳过前 offset 行再按 limit 收集
Key string `json:"key"` // Redis key (get/delete)
Pattern string `json:"pattern"` // Redis SCAN match
Cursor int64 `json:"cursor"` // Redis 值分页游标(hash/set=SCAN 游标;list/zset=偏移)
Keyword string `json:"keyword"` // 对象搜索关键词
Format string `json:"format"` // 导出格式: csv | json | insert
Offset int `json:"offset"` // 结果分页:跳过前 offset 行再按 limit 收集
Key string `json:"key"` // Redis key (get/delete)
Pattern string `json:"pattern"` // Redis SCAN match
KeyCursor string `json:"keyCursor"` // Redis 键列表的跨主节点续扫位置
Cursor int64 `json:"cursor"` // Redis 值分页游标(hash/set=SCAN 游标;list/zset=偏移)
Keyword string `json:"keyword"` // 对象搜索关键词
Format string `json:"format"` // 导出格式: csv | json | insert
// TimeoutMs > 0 时 /db/exec、/db/explain 以此为语句超时
TimeoutMs int `json:"timeoutMs"`
@@ -77,6 +78,7 @@ type ColumnInfo struct {
Key string `json:"key,omitempty"`
Default any `json:"default,omitempty"`
Comment string `json:"comment,omitempty"`
Extra string `json:"extra,omitempty"`
}
// ExecResult 与前端 QueryResult 字段对齐
@@ -129,29 +131,32 @@ func normalize(v interface{}) interface{} {
// Redis key browse (dedicated endpoints; do not reshape SQL tables []).
type RedisKeyInfo struct {
Name string `json:"name"`
Type string `json:"type"`
TTL int64 `json:"ttl"` // seconds; -1 no expire; -2 missing
Name string `json:"name"`
Type string `json:"type"`
TTL int64 `json:"ttl"` // seconds; -1 no expire; -2 missing
Exact bool `json:"exact,omitempty"` // TYPE 精确命中置顶
}
type RedisKeysResult struct {
Keys []RedisKeyInfo `json:"keys"`
Truncated bool `json:"truncated,omitempty"`
TimeMs int64 `json:"timeMs"`
Error string `json:"error,omitempty"`
Keys []RedisKeyInfo `json:"keys"`
NextCursor string `json:"nextCursor,omitempty"`
ExactHit bool `json:"exactHit,omitempty"` // 无通配符时 TYPE 直接命中
Truncated bool `json:"truncated,omitempty"`
TimeMs int64 `json:"timeMs"`
Error string `json:"error,omitempty"`
}
// RedisKeyValue feeds ResultPane (columns/rows) plus type/ttl meta.
type RedisKeyValue struct {
Name string `json:"name"`
Type string `json:"type"`
TTL int64 `json:"ttl"`
Columns []string `json:"columns"`
Rows []map[string]interface{} `json:"rows"`
RowCount int `json:"rowCount"`
TimeMs int64 `json:"timeMs"`
Truncated bool `json:"truncated,omitempty"`
Limit int `json:"limit,omitempty"`
NextCursor int64 `json:"nextCursor"` // >0 表示还有更多(续载时回传 cursor)
Error string `json:"error,omitempty"`
Name string `json:"name"`
Type string `json:"type"`
TTL int64 `json:"ttl"`
Columns []string `json:"columns"`
Rows []map[string]interface{} `json:"rows"`
RowCount int `json:"rowCount"`
TimeMs int64 `json:"timeMs"`
Truncated bool `json:"truncated,omitempty"`
Limit int `json:"limit,omitempty"`
NextCursor int64 `json:"nextCursor"` // >0 表示还有更多(续载时回传 cursor)
Error string `json:"error,omitempty"`
}