feat: MySQL/PG 实时指标采集 + 版本检测修复 + 密码加密存储

- 新增 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 函数
This commit is contained in:
logan committed 2026-05-31 00:15:18 +08:00
1 parent 433e7e2909
commit 31db229d70
2 files changed
+184 -3

No files matched your search

+164 -2
View File
@@ -4,6 +4,7 @@ import (
"context" "context"
"database/sql" "database/sql"
"fmt" "fmt"
"strconv"
"strings" "strings"
"time" "time"
@@ -83,8 +84,12 @@ func extractVersion(msg, prefix string) string {
return "" return ""
} }
ver := msg[idx+len(prefix):] ver := msg[idx+len(prefix):]
if end := strings.Index(ver, ")"); end > 0 { // 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] ver = ver[:end]
break
}
} }
return strings.TrimSpace(ver) return strings.TrimSpace(ver)
} }
@@ -280,7 +285,164 @@ func getRedisMetrics(instance *model.Instance) map[string]interface{} {
return metrics 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) { func getDBConnection(instance *model.Instance, dbName string) (*sql.DB, error) {
password, err := decryptPassword(instance.Password) password, err := decryptPassword(instance.Password)
if err != nil { if err != nil {
+20 -1
View File
@@ -64,6 +64,13 @@ func CreateInstance(c *gin.Context) {
c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
return 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 { if err := model.DB.Create(&instance).Error; err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()}) c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return return
@@ -105,6 +112,13 @@ func UpdateInstance(c *gin.Context) {
if pw, ok := input["password"].(string); ok && pw == "******" { if pw, ok := input["password"].(string); ok && pw == "******" {
delete(input, "password") 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 { if err := model.DB.Model(&instance).Updates(input).Error; err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()}) c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
@@ -191,8 +205,13 @@ func InstanceMetrics(c *gin.Context) {
case "redis": case "redis":
metrics := getRedisMetrics(&instance) metrics := getRedisMetrics(&instance)
c.JSON(http.StatusOK, metrics) 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: default:
// For MySQL/PG: return basic status
c.JSON(http.StatusOK, gin.H{ c.JSON(http.StatusOK, gin.H{
"status": "online", "status": "online",
"version": instance.Version, "version": instance.Version,