feat: 构建节点磁盘性能测试（dd 实测读写）
- 新增 internal/sshexec（x/crypto/ssh）：远程命令通道，凭证来自 Credential(type=ssh) 解密（私钥/密码/端口/用户） - POST /api/build-nodes/:id/disk-test：写 dd(256MB, oflag=direct， 不支持自动降级含缓存并标注) + 读 dd + 清理；IP 为本机时直接本地 exec；输出解析兼容 Linux/Mac 两种 dd 格式；60s 超时保护 - BuildNode 新增 credentialId 关联 SSH 凭证 + 磁盘测速结果持久字段 （显式 column 标签修正 GORM MBps 命名） - 节点卡片新增仪表盘按钮触发测试；磁盘行显示 ↑写 ↓读 MB/s （>200 绿 SSD 级 / 50-200 黄 / <50 红），hover 看测试模式与时间 - 编辑弹窗新增 SSH 凭证下拉 - 本机实测：local-mac 节点 写 2336 MB/s · 读 1793 MB/s（direct）
This commit is contained in:
Logan.Li committed
1 parent
6841c1104a
commit
555934cc64
8 files changed
+314
-2
No files matched your search
@@ -0,0 +1,176 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os/exec"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"flowstack/internal/crypto"
|
||||
"flowstack/internal/database"
|
||||
"flowstack/internal/model"
|
||||
"flowstack/internal/sshexec"
|
||||
)
|
||||
|
||||
const (
|
||||
ddFileSize = ".fs-disktest"
|
||||
ddSizeMB = 256
|
||||
ddTimeout = 60 * time.Second
|
||||
)
|
||||
|
||||
type cmdRunner func(cmd string) (string, error)
|
||||
|
||||
func localRunner(cmd string) (string, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), ddTimeout)
|
||||
defer cancel()
|
||||
out, err := exec.CommandContext(ctx, "sh", "-c", cmd).CombinedOutput()
|
||||
return string(out), err
|
||||
}
|
||||
|
||||
// parseDDMBps 兼容两种 dd 输出：
|
||||
// Linux: "268435456 bytes (268 MB, 256 MiB) copied, 1.2 s, 224 MB/s"
|
||||
// macOS: "268435456 bytes transferred in 1.234 secs (217543210 bytes/sec)"
|
||||
var (
|
||||
linuxSpeedRe = regexp.MustCompile(`([\d.]+)\s*([kMGT]?)B/s`)
|
||||
macSpeedRe = regexp.MustCompile(`\(([\d,]+)\s*bytes/sec\)`)
|
||||
)
|
||||
|
||||
func parseDDMBps(out string) float64 {
|
||||
if ms := linuxSpeedRe.FindAllStringSubmatch(out, -1); len(ms) > 0 {
|
||||
m := ms[len(ms)-1]
|
||||
v, err := strconv.ParseFloat(m[1], 64)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
switch m[2] {
|
||||
case "k", "K":
|
||||
v /= 1024
|
||||
case "G":
|
||||
v *= 1024
|
||||
case "T":
|
||||
v *= 1024 * 1024
|
||||
}
|
||||
return v
|
||||
}
|
||||
if m := macSpeedRe.FindStringSubmatch(out); m != nil {
|
||||
v, err := strconv.ParseFloat(strings.ReplaceAll(m[1], ",", ""), 64)
|
||||
if err == nil {
|
||||
return v / 1e6
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// TestBuildNodeDisk POST /api/build-nodes/:id/disk-test —— dd 实测磁盘读写
|
||||
func TestBuildNodeDisk(c *gin.Context) {
|
||||
var n model.BuildNode
|
||||
if err := database.DB.First(&n, c.Param("id")).Error; err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "build node not found"})
|
||||
return
|
||||
}
|
||||
|
||||
// 本地节点直接 exec，否则 SSH（需要凭证）
|
||||
isLocal := n.IP == "127.0.0.1" || n.IP == "localhost" || n.IP == "::1"
|
||||
var run cmdRunner
|
||||
mode := "ssh"
|
||||
if isLocal {
|
||||
run = localRunner
|
||||
mode = "local"
|
||||
} else {
|
||||
if n.CredentialID == nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "远程节点需先在编辑中关联 SSH 凭证"})
|
||||
return
|
||||
}
|
||||
var cred model.Credential
|
||||
if err := database.DB.First(&cred, *n.CredentialID).Error; err != nil || cred.Content == "" {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "SSH 凭证不存在或未设置内容"})
|
||||
return
|
||||
}
|
||||
plain, err := crypto.DecryptField(cred.Content)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "凭证解密失败"})
|
||||
return
|
||||
}
|
||||
var cc struct {
|
||||
PrivateKey string `json:"private_key"`
|
||||
Password string `json:"password"`
|
||||
SSHUser string `json:"ssh_user"`
|
||||
SSHPort int `json:"ssh_port"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(plain), &cc); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "凭证内容格式错误"})
|
||||
return
|
||||
}
|
||||
cfg := sshexec.Config{Host: n.IP, Port: cc.SSHPort, User: cc.SSHUser, Password: cc.Password, PrivateKey: cc.PrivateKey}
|
||||
run = func(cmd string) (string, error) {
|
||||
return sshexec.Run(cfg, cmd, ddTimeout)
|
||||
}
|
||||
}
|
||||
|
||||
ws := n.WorkspacePath
|
||||
if ws == "" {
|
||||
ws = "/data/flowstack-workspace"
|
||||
}
|
||||
file := ws + "/" + ddFileSize
|
||||
start := time.Now()
|
||||
|
||||
if _, err := run(fmt.Sprintf("mkdir -p '%s'", ws)); err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{"success": false, "message": "创建工作空间失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
defer func() { _, _ = run(fmt.Sprintf("rm -f '%s'", file)) }()
|
||||
|
||||
// 写测试：direct 优先（绕过缓存），失败降级普通模式
|
||||
writeCmd := fmt.Sprintf("dd if=/dev/zero of='%s' bs=1m count=%d oflag=direct 2>&1", file, ddSizeMB)
|
||||
writeOut, err := run(writeCmd)
|
||||
direct := err == nil
|
||||
if err != nil || parseDDMBps(writeOut) == 0 {
|
||||
writeOut, err = run(fmt.Sprintf("dd if=/dev/zero of='%s' bs=1m count=%d 2>&1", file, ddSizeMB))
|
||||
direct = false
|
||||
}
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{"success": false, "message": "写入测试失败: " + firstLine(writeOut, err)})
|
||||
return
|
||||
}
|
||||
writeMBps := parseDDMBps(writeOut)
|
||||
|
||||
// 读测试
|
||||
readCmd := fmt.Sprintf("dd if='%s' of=/dev/null bs=1m 2>&1", file)
|
||||
if direct {
|
||||
readCmd = fmt.Sprintf("dd if='%s' of=/dev/null bs=1m iflag=direct 2>&1", file)
|
||||
}
|
||||
readOut, err := run(readCmd)
|
||||
if err != nil && direct {
|
||||
readOut, err = run(fmt.Sprintf("dd if='%s' of=/dev/null bs=1m 2>&1", file))
|
||||
}
|
||||
if err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{"success": false, "message": "读取测试失败: " + firstLine(readOut, err)})
|
||||
return
|
||||
}
|
||||
readMBps := parseDDMBps(readOut)
|
||||
|
||||
if direct {
|
||||
mode += "/direct"
|
||||
} else {
|
||||
mode += "/cache"
|
||||
}
|
||||
|
||||
// 回写节点
|
||||
now := time.Now()
|
||||
database.DB.Model(&n).Updates(map[string]any{
|
||||
"disk_write_mbps": writeMBps, "disk_read_mbps": readMBps,
|
||||
"disk_test_mode": mode, "disk_tested_at": &now,
|
||||
})
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"success": true, "writeMBps": writeMBps, "readMBps": readMBps,
|
||||
"mode": mode, "elapsedMs": time.Since(start).Milliseconds(),
|
||||
"message": fmt.Sprintf("写入 %.0f MB/s · 读取 %.0f MB/s（%s）", writeMBps, readMBps, map[bool]string{true: "direct 直测", false: "含缓存"}[direct]),
|
||||
})
|
||||
}
|
||||
@@ -394,7 +394,7 @@ func UpdateBuildNode(c *gin.Context) {
|
||||
}
|
||||
var raw map[string]any
|
||||
_ = c.ShouldBindJSON(&raw)
|
||||
for _, k := range []string{"name", "ip", "labels", "maxConcurrency", "workspacePath", "diskInfo", "enabled"} {
|
||||
for _, k := range []string{"name", "ip", "labels", "maxConcurrency", "workspacePath", "diskInfo", "credentialId", "enabled"} {
|
||||
if v, ok := raw[k]; ok {
|
||||
database.DB.Model(&n).Update(k, v)
|
||||
}
|
||||
|
||||
@@ -205,6 +205,11 @@ type BuildNode struct {
|
||||
MaxConcurrency int `gorm:"default:2" json:"maxConcurrency"`
|
||||
WorkspacePath string `gorm:"default:/data/flowstack-workspace" json:"workspacePath"`
|
||||
DiskInfo string `json:"diskInfo"` // 如 "45G/200G"
|
||||
CredentialID *uint `json:"credentialId"` // SSH 凭证（磁盘性能测试等远程操作）
|
||||
DiskWriteMBps float64 `gorm:"column:disk_write_mbps" json:"diskWriteMBps"`
|
||||
DiskReadMBps float64 `gorm:"column:disk_read_mbps" json:"diskReadMBps"`
|
||||
DiskTestMode string `json:"diskTestMode"` // local|ssh + /direct|/cache
|
||||
DiskTestedAt *time.Time `json:"diskTestedAt"`
|
||||
Enabled bool `gorm:"default:true" json:"enabled"`
|
||||
Status string `gorm:"default:offline" json:"status"` // online|offline|busy
|
||||
CurrentJobs int `gorm:"default:0" json:"currentJobs"`
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
// Package sshexec 远程命令执行通道（构建节点 SSH 操作）。
|
||||
package sshexec
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// Config SSH 连接参数（来自 SSH 凭证内容 JSON）
|
||||
type Config struct {
|
||||
Host string // IP
|
||||
Port int // 默认 22
|
||||
User string
|
||||
Password string
|
||||
PrivateKey string // PEM 私钥（与 Password 二选一）
|
||||
}
|
||||
|
||||
// Run 执行单条远程命令，返回合并输出；timeout 为整条命令的硬上限
|
||||
func Run(cfg Config, cmd string, timeout time.Duration) (string, error) {
|
||||
if cfg.Port == 0 {
|
||||
cfg.Port = 22
|
||||
}
|
||||
if cfg.User == "" {
|
||||
cfg.User = "root"
|
||||
}
|
||||
|
||||
var authMethods []ssh.AuthMethod
|
||||
if cfg.PrivateKey != "" {
|
||||
signer, err := ssh.ParsePrivateKey([]byte(cfg.PrivateKey))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("私钥解析失败: %v", err)
|
||||
}
|
||||
authMethods = append(authMethods, ssh.PublicKeys(signer))
|
||||
}
|
||||
if cfg.Password != "" {
|
||||
authMethods = append(authMethods, ssh.Password(cfg.Password))
|
||||
}
|
||||
if len(authMethods) == 0 {
|
||||
return "", fmt.Errorf("未提供密码或私钥")
|
||||
}
|
||||
|
||||
client, err := ssh.Dial("tcp", net.JoinHostPort(cfg.Host, fmt.Sprint(cfg.Port)), &ssh.ClientConfig{
|
||||
User: cfg.User,
|
||||
Auth: authMethods,
|
||||
Timeout: 10 * time.Second,
|
||||
HostKeyCallback: ssh.InsecureIgnoreHostKey(), // 内网构建节点场景
|
||||
})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("SSH 连接失败: %v", err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
session, err := client.NewSession()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer session.Close()
|
||||
|
||||
done := make(chan struct{})
|
||||
var output []byte
|
||||
var runErr error
|
||||
go func() {
|
||||
defer close(done)
|
||||
output, runErr = session.CombinedOutput(cmd)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
return string(output), runErr
|
||||
case <-time.After(timeout):
|
||||
_ = session.Close()
|
||||
return string(output), fmt.Errorf("命令执行超时（%s）", timeout)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user