From 31db229d70fd63f82be91f67850590fa9c9fb31a Mon Sep 17 00:00:00 2001 From: logan Date: Sun, 31 May 2026 00:15:18 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20MySQL/PG=20=E5=AE=9E=E6=97=B6=E6=8C=87?= =?UTF-8?q?=E6=A0=87=E9=87=87=E9=9B=86=20+=20=E7=89=88=E6=9C=AC=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E4=BF=AE=E5=A4=8D=20+=20=E5=AF=86=E7=A0=81=E5=8A=A0?= =?UTF-8?q?=E5=AF=86=E5=AD=98=E5=82=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 getMySQLMetrics(): SHOW GLOBAL STATUS 采集连接数/QPS/延迟/运行时间 - 新增 getPostgresMetrics(): pg_stat_activity 采集活跃连接+运行时间 - 修复 extractVersion(): 支持 Redis 'version: 7.4.9, mode: standalone' 格式(逗号截断) - CreateInstance/UpdateInstance: 创建/更新时自动加密密码(enc: 前缀) - InstanceMetrics handler: MySQL/PG 分别调用专属 metrics 函数 --- backend/internal/handler/connection_tester.go | 168 +++++++++++++++++- backend/internal/handler/instance.go | 21 ++- 2 files changed, 185 insertions(+), 4 deletions(-) diff --git a/backend/internal/handler/connection_tester.go b/backend/internal/handler/connection_tester.go index 167957c..07324c6 100644 --- a/backend/internal/handler/connection_tester.go +++ b/backend/internal/handler/connection_tester.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "strconv" "strings" "time" @@ -83,8 +84,12 @@ func extractVersion(msg, prefix string) string { return "" } ver := msg[idx+len(prefix):] - if end := strings.Index(ver, ")"); end > 0 { - ver = ver[:end] + // Cut at first comma, closing paren, or space (whichever comes first) + for _, sep := range []string{")", ",", " "} { + if end := strings.Index(ver, sep); end > 0 { + ver = ver[:end] + break + } } return strings.TrimSpace(ver) } @@ -280,7 +285,164 @@ func getRedisMetrics(instance *model.Instance) map[string]interface{} { return metrics } -// getDB opens a real database connection and returns *sql.DB +// getMySQLMetrics fetches live MySQL metrics (connections, QPS, latency, uptime, version) +func getMySQLMetrics(instance *model.Instance) map[string]interface{} { + password := instance.Password + if strings.HasPrefix(password, "enc:") { + decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:")) + if err == nil { + password = decrypted + } + } + + metrics := map[string]interface{}{ + "status": "offline", + "connections": 0, + "qps": 0, + "latency": 0, + "uptime": "-", + "version": "", + } + if instance.Version != nil { + metrics["version"] = *instance.Version + } + + dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/?timeout=5s&parseTime=true", + instance.Username, password, instance.Host, instance.Port) + db, err := sql.Open("mysql", dsn) + if err != nil { + return metrics + } + defer db.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Latency from simple query + start := time.Now() + if err := db.PingContext(ctx); err != nil { + return metrics + } + metrics["latency"] = time.Since(start).Milliseconds() + metrics["status"] = "online" + + // Fetch STATUS variables + statusVars := map[string]string{} + rows, err := db.QueryContext(ctx, "SHOW GLOBAL STATUS WHERE Variable_name IN "+ + "('Threads_connected','Queries','Uptime','Connections','Slow_queries')") + if err == nil { + defer rows.Close() + for rows.Next() { + var name, value string + rows.Scan(&name, &value) + statusVars[name] = value + } + } + + if v, ok := statusVars["Threads_connected"]; ok { + n, _ := strconv.Atoi(v) + metrics["connections"] = n + } + if v, ok := statusVars["Uptime"]; ok { + secs, _ := strconv.Atoi(v) + if secs > 86400 { + metrics["uptime"] = fmt.Sprintf("%d 天", secs/86400) + } else if secs > 3600 { + metrics["uptime"] = fmt.Sprintf("%d 小时", secs/3600) + } else { + metrics["uptime"] = fmt.Sprintf("%d 分钟", secs/60) + } + } + // QPS = Queries / Uptime + if q, ok := statusVars["Queries"]; ok { + if u, ok2 := statusVars["Uptime"]; ok2 { + qi, _ := strconv.ParseFloat(q, 64) + ui, _ := strconv.ParseFloat(u, 64) + if ui > 0 { + metrics["qps"] = int(qi / ui) + } + } + } + + // Version from DB if not already set + if instance.Version == nil || *instance.Version == "" { + var ver string + if err := db.QueryRowContext(ctx, "SELECT VERSION()").Scan(&ver); err == nil { + metrics["version"] = ver + } + } + + return metrics +} + +// getPostgresMetrics fetches live PostgreSQL metrics +func getPostgresMetrics(instance *model.Instance) map[string]interface{} { + password := instance.Password + if strings.HasPrefix(password, "enc:") { + decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:")) + if err == nil { + password = decrypted + } + } + + metrics := map[string]interface{}{ + "status": "offline", + "connections": 0, + "qps": 0, + "latency": 0, + "uptime": "-", + "version": "", + } + if instance.Version != nil { + metrics["version"] = *instance.Version + } + + dsn := fmt.Sprintf("postgres://%s:%s@%s:%d/postgres?sslmode=disable&connect_timeout=5", + instance.Username, password, instance.Host, instance.Port) + db, err := sql.Open("pgx", dsn) + if err != nil { + return metrics + } + defer db.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + start := time.Now() + if err := db.PingContext(ctx); err != nil { + return metrics + } + metrics["latency"] = time.Since(start).Milliseconds() + metrics["status"] = "online" + + // Fetch pg_stat_activity counts + var activeConns int + db.QueryRowContext(ctx, "SELECT count(*) FROM pg_stat_activity WHERE state = 'active'").Scan(&activeConns) + metrics["connections"] = activeConns + + // Uptime + var uptimeSecs int + if err := db.QueryRowContext(ctx, "SELECT extract(epoch from now() - pg_postmaster_start_time())::int").Scan(&uptimeSecs); err == nil { + if uptimeSecs > 86400 { + metrics["uptime"] = fmt.Sprintf("%d 天", uptimeSecs/86400) + } else if uptimeSecs > 3600 { + metrics["uptime"] = fmt.Sprintf("%d 小时", uptimeSecs/3600) + } else { + metrics["uptime"] = fmt.Sprintf("%d 分钟", uptimeSecs/60) + } + } + + // Version + if instance.Version == nil || *instance.Version == "" { + var ver string + if err := db.QueryRowContext(ctx, "SELECT version()").Scan(&ver); err == nil { + metrics["version"] = ver + } + } + + return metrics +} + func getDBConnection(instance *model.Instance, dbName string) (*sql.DB, error) { password, err := decryptPassword(instance.Password) if err != nil { diff --git a/backend/internal/handler/instance.go b/backend/internal/handler/instance.go index 0e358aa..af65dab 100644 --- a/backend/internal/handler/instance.go +++ b/backend/internal/handler/instance.go @@ -64,6 +64,13 @@ func CreateInstance(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) return } + // Encrypt password before storing + if instance.Password != "" { + encPw, err := encryptPassword(instance.Password) + if err == nil { + instance.Password = "enc:" + encPw + } + } if err := model.DB.Create(&instance).Error; err != nil { c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()}) return @@ -105,6 +112,13 @@ func UpdateInstance(c *gin.Context) { if pw, ok := input["password"].(string); ok && pw == "******" { delete(input, "password") } + // Encrypt password if provided and not masked + if pw, ok := input["password"].(string); ok && pw != "" { + encPw, err := encryptPassword(pw) + if err == nil { + input["password"] = "enc:" + encPw + } + } if err := model.DB.Model(&instance).Updates(input).Error; err != nil { c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()}) @@ -191,8 +205,13 @@ func InstanceMetrics(c *gin.Context) { case "redis": metrics := getRedisMetrics(&instance) c.JSON(http.StatusOK, metrics) + case "mysql": + metrics := getMySQLMetrics(&instance) + c.JSON(http.StatusOK, metrics) + case "postgresql", "postgres", "pg": + metrics := getPostgresMetrics(&instance) + c.JSON(http.StatusOK, metrics) default: - // For MySQL/PG: return basic status c.JSON(http.StatusOK, gin.H{ "status": "online", "version": instance.Version,