diff --git a/backend/go.sum b/backend/go.sum index dcd7994..04651fc 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -96,6 +96,8 @@ golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/term v0.33.0 h1:NuFncQrRcaRvVmgRkvM3j/F00gWIAlcmlB8ACEKmGIg= +golang.org/x/term v0.33.0/go.mod h1:s18+ql9tYWp1IfpV9DmCtQDDSRBUjKaw9M1eAv5UeF0= golang.org/x/text v0.27.0 h1:4fGWRpyh641NLlecmyl4LOe6yDdfaYNrGb2zdfo4JV4= golang.org/x/text v0.27.0/go.mod h1:1D28KMCvyooCX9hBiosv5Tz/+YLxj0j7XhWjpSUF7CU= golang.org/x/tools v0.34.0 h1:qIpSLOxeCYGg9TrcJokLBG4KFA6d795g0xkBkiESGlo= diff --git a/backend/internal/handler/disktest.go b/backend/internal/handler/disktest.go new file mode 100644 index 0000000..99116a4 --- /dev/null +++ b/backend/internal/handler/disktest.go @@ -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]), + }) +} diff --git a/backend/internal/handler/resources.go b/backend/internal/handler/resources.go index 3c2deeb..10496c3 100644 --- a/backend/internal/handler/resources.go +++ b/backend/internal/handler/resources.go @@ -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) } diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 81be180..af702fc 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -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"` diff --git a/backend/internal/sshexec/client.go b/backend/internal/sshexec/client.go new file mode 100644 index 0000000..e39813d --- /dev/null +++ b/backend/internal/sshexec/client.go @@ -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) + } +} diff --git a/backend/main.go b/backend/main.go index 461db8f..ba4bfb5 100644 --- a/backend/main.go +++ b/backend/main.go @@ -110,6 +110,7 @@ func main() { api.PUT("/build-nodes/:id", handler.UpdateBuildNode) api.DELETE("/build-nodes/:id", handler.DeleteBuildNode) api.POST("/build-nodes/:id/test", handler.TestBuildNode) + api.POST("/build-nodes/:id/disk-test", handler.TestBuildNodeDisk) api.GET("/tools", handler.ListTools) api.PUT("/tools/:id", handler.UpdateTool) diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index e4c4947..e475fa5 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -125,6 +125,7 @@ export const apiClient = { updateBuildNode: (id: any, b: any) => api(`/api/build-nodes/${id}`, { method: 'PUT', body: JSON.stringify(b) }), deleteBuildNode: (id: any) => api(`/api/build-nodes/${id}`, { method: 'DELETE' }), testBuildNode: (id: any) => api(`/api/build-nodes/${id}/test`, { method: 'POST' }), + testBuildNodeDisk: (id: any) => api(`/api/build-nodes/${id}/disk-test`, { method: 'POST' }), // 工具 listTools: () => api('/api/tools'), updateTool: (id: any, b: any) => api(`/api/tools/${id}`, { method: 'PUT', body: JSON.stringify(b) }), diff --git a/frontend/src/pages/BuildNodeManager.tsx b/frontend/src/pages/BuildNodeManager.tsx index 4be2425..828ad5c 100644 --- a/frontend/src/pages/BuildNodeManager.tsx +++ b/frontend/src/pages/BuildNodeManager.tsx @@ -11,6 +11,7 @@ import { Tag, Zap, HardDrive, + Gauge, } from 'lucide-react'; import { toast } from 'sonner'; import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; @@ -77,6 +78,9 @@ export default function BuildNodeManager() { const [searchQuery, setSearchQuery] = useState(''); const nodesQuery = useQuery({ queryKey: ['build-nodes'], queryFn: () => apiClient.listBuildNodes() }); + // SSH 凭证（磁盘性能测试等远程操作用） + const credentialsQuery = useQuery({ queryKey: ['credentials', 'ssh'], queryFn: () => apiClient.listCredentials('ssh') }); + const sshCredentials: any[] = credentialsQuery.data ?? []; const nodes: any[] = nodesQuery.data ?? []; // ---- 弹窗 ---- @@ -133,6 +137,17 @@ export default function BuildNodeManager() { onError: (err: Error) => toast.error(err.message), }); + // 磁盘性能测试（dd 实测） + const diskMutation = useMutation({ + mutationFn: (id: any) => apiClient.testBuildNodeDisk(id), + onSuccess: (result: any) => { + if (result?.success) toast.success('磁盘性能：' + result.message, { duration: 6000 }); + else toast.error(result?.message || '磁盘测试失败'); + invalidate(); + }, + onError: (err: Error) => toast.error('磁盘测试失败', { description: err.message }), + }); + const toggleMutation = useMutation({ mutationFn: (vars: { id: any; enabled: boolean }) => apiClient.updateBuildNode(vars.id, { enabled: vars.enabled }), onSuccess: (_data, vars) => { @@ -152,6 +167,7 @@ export default function BuildNodeManager() { body: { name: form.name.trim(), ip: form.ip.trim(), + credentialId: editing?.credentialId ?? null, labels: JSON.stringify(splitComma(form.labels)), maxConcurrency: Number(form.maxConcurrency) || 1, workspacePath: form.workspacePath.trim(), @@ -252,6 +268,18 @@ export default function BuildNodeManager() { onCheckedChange={(checked) => toggleMutation.mutate({ id: node.id, enabled: checked })} /> +