diff --git a/backend/internal/crypto/crypto.go b/backend/internal/crypto/crypto.go index d62efcb..8dace9d 100644 --- a/backend/internal/crypto/crypto.go +++ b/backend/internal/crypto/crypto.go @@ -1,15 +1,132 @@ -// Package crypto 提供口令哈希原语（argon2id，与 opsx-loki 同参数）。 +// Package crypto 提供口令哈希（argon2id）与字段加密（AES-256-GCM，enc:v1: 前缀）原语。 package crypto import ( + "crypto/aes" + "crypto/cipher" "crypto/rand" "crypto/subtle" "encoding/base64" + "errors" + "io" "math/big" + "os" + "path/filepath" + "strings" + "sync" "golang.org/x/crypto/argon2" ) +const encPrefix = "enc:v1:" + +var ( + encKey []byte + encKeyOnce sync.Once + encKeyErr error +) + +// InitEncryptionKey 初始化字段加密密钥： +// 优先环境变量 FLOWSTACK_ENCRYPTION_KEY，否则首次启动生成 32 字节随机密钥 +// 写入 dataDir/encryption.key（0600），重启复用。 +func InitEncryptionKey(dataDir string) error { + encKeyOnce.Do(func() { + if env := os.Getenv("FLOWSTACK_ENCRYPTION_KEY"); env != "" { + encKey = []byte(env) + return + } + keyFile := filepath.Join(dataDir, "encryption.key") + if data, err := os.ReadFile(keyFile); err == nil && len(data) == 32 { + encKey = data + return + } + key := make([]byte, 32) + if _, err := rand.Read(key); err != nil { + encKeyErr = err + return + } + if err := os.WriteFile(keyFile, key, 0o600); err != nil { + encKeyErr = err + return + } + encKey = key + }) + return encKeyErr +} + +func gcm() (cipher.AEAD, error) { + if encKey == nil { + return nil, errors.New("加密密钥未初始化") + } + block, err := aes.NewCipher(encKey) + if err != nil { + return nil, err + } + return cipher.NewGCM(block) +} + +// EncryptField AES-256-GCM 加密，返回 enc:v1:base64；空串原样返回 +func EncryptField(plaintext string) (string, error) { + if plaintext == "" { + return "", nil + } + aead, err := gcm() + if err != nil { + return "", err + } + nonce := make([]byte, aead.NonceSize()) + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return "", err + } + ciphertext := aead.Seal(nonce, nonce, []byte(plaintext), nil) + return encPrefix + base64.StdEncoding.EncodeToString(ciphertext), nil +} + +// DecryptField 解密 enc:v1: 前缀密文；非该格式原样返回（向后兼容） +func DecryptField(encrypted string) (string, error) { + if !strings.HasPrefix(encrypted, encPrefix) { + return encrypted, nil + } + data, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(encrypted, encPrefix)) + if err != nil { + return "", err + } + aead, err := gcm() + if err != nil { + return "", err + } + if len(data) < aead.NonceSize() { + return "", errors.New("ciphertext too short") + } + nonce, ciphertext := data[:aead.NonceSize()], data[aead.NonceSize():] + plain, err := aead.Open(nil, nonce, ciphertext, nil) + if err != nil { + return "", err + } + return string(plain), nil +} + +// IsEncrypted 判断字段是否为加密格式 +func IsEncrypted(v string) bool { + return strings.HasPrefix(v, encPrefix) +} + +// Mask 敏感值脱敏展示（保留前后各 1/4，中间星号，最长 24 位） +func Mask(s string) string { + if s == "" { + return "" + } + r := []rune(s) + if len(r) <= 4 { + return strings.Repeat("*", len(r)) + } + keep := len(r) / 4 + if keep > 4 { + keep = 4 + } + return string(r[:keep]) + strings.Repeat("*", 6) + string(r[len(r)-keep:]) +} + // HashPassword 生成 argon2id 口令哈希（salt+hash base64 编码） func HashPassword(password string) (string, error) { salt := make([]byte, 16) diff --git a/backend/internal/database/database.go b/backend/internal/database/database.go index 0f45ac0..e67ca6e 100644 --- a/backend/internal/database/database.go +++ b/backend/internal/database/database.go @@ -4,6 +4,7 @@ import ( "log" "os" "path/filepath" + "time" "github.com/glebarez/sqlite" "gorm.io/gorm" @@ -23,6 +24,9 @@ func Init() error { if err := os.MkdirAll(DataDir, 0o755); err != nil { return err } + if err := crypto.InitEncryptionKey(DataDir); err != nil { + return err + } db, err := gorm.Open(sqlite.Open(filepath.Join(DataDir, "flowstack.db")), &gorm.Config{}) if err != nil { @@ -35,12 +39,16 @@ func Init() error { &model.Pipeline{}, &model.PipelineNode{}, &model.PipelineEdge{}, &model.PipelineRun{}, &model.RunStage{}, &model.LogEntry{}, &model.Application{}, &model.Plugin{}, + &model.Host{}, &model.K8sCluster{}, &model.Credential{}, &model.BuildNode{}, + &model.DeployTool{}, &model.WebhookEndpoint{}, &model.JenkinsServer{}, &model.SystemSetting{}, ); err != nil { return err } seedAdmin() seedDemoData() + seedResources() + seedTools() return nil } @@ -153,9 +161,10 @@ func seedDemoData() { }) // ---- GitOps 应用 ---- + now := time.Now() DB.Create([]model.Application{ - {Name: "aiops-backend", Repo: "https://gitea.91ai.io/rlwj-lab/aiops-gitops.git", Branch: "main", Path: "apps/aiops-backend", Cluster: "ack-production", Namespace: "aiops", ManifestType: "kustomize", SyncPolicy: "automated", AutoPrune: true, SelfHeal: true}, - {Name: "aiops-web", Repo: "https://gitea.91ai.io/rlwj-lab/aiops-gitops.git", Branch: "main", Path: "apps/aiops-web", Cluster: "ack-production", Namespace: "aiops", ManifestType: "kustomize", SyncPolicy: "manual"}, + {Name: "aiops-backend", Repo: "https://gitea.91ai.io/rlwj-lab/aiops-gitops.git", Branch: "main", Path: "apps/aiops-backend", Cluster: "ack-production", Namespace: "aiops", ManifestType: "kustomize", SyncPolicy: "automated", AutoPrune: true, SelfHeal: true, SyncStatus: "Synced", HealthStatus: "Healthy", LastSyncAt: &now}, + {Name: "aiops-web", Repo: "https://gitea.91ai.io/rlwj-lab/aiops-gitops.git", Branch: "main", Path: "apps/aiops-web", Cluster: "ack-production", Namespace: "aiops", ManifestType: "kustomize", SyncPolicy: "manual", SyncStatus: "OutOfSync", HealthStatus: "Progressing"}, }) // ---- 已安装插件 ---- @@ -165,3 +174,69 @@ func seedDemoData() { {PluginID: "k8s-rollout", Name: "K8s 滚动发布", Description: "Deployment 滚动更新与回滚", Version: "2.3.1", Author: "CloudNative", Category: "构建", Enabled: false}, }) } + +// seedResources 资源示例数据（主机/集群/凭证/构建节点/Webhook，各表独立判空） +func seedResources() { + var n int64 + DB.Model(&model.Host{}).Count(&n) + if n == 0 { + DB.Create([]model.Host{ + {Name: "build-host-01", IP: "10.0.8.11", SSHPort: 22, Username: "root", Tags: `["build","docker"]`, Status: "online", OS: "Ubuntu 22.04", Remark: "通用构建机"}, + {Name: "prod-web-01", IP: "10.0.1.21", SSHPort: 22, Username: "admin", Tags: `["production","web"]`, Status: "online", OS: "CentOS 7.9", Remark: "Nginx 前端"}, + }) + } + DB.Model(&model.K8sCluster{}).Count(&n) + if n == 0 { + DB.Create([]model.K8sCluster{ + {Name: "ack-production", Context: "ack-production", APIServer: "https://k8s.prod:6443", Version: "v1.30.1", NodeCount: 3, Namespaces: `["aiops","default","kube-system"]`, Status: "disconnected"}, + }) + } + DB.Model(&model.Credential{}).Count(&n) + if n == 0 { + enc1, _ := crypto.EncryptField(`{"private_key":"CHANGE_ME","ssh_user":"root","ssh_port":22}`) + DB.Create([]model.Credential{ + {Name: "prod-ssh-key", Type: "ssh", Scope: "production-group", Content: enc1, Summary: "ssh_key root@22"}, + {Name: "registry-example", Type: "docker", Scope: "全局", Summary: "docker registry.example.com aiops-pull"}, + }) + } + DB.Model(&model.BuildNode{}).Count(&n) + if n == 0 { + hb := time.Now().Add(-10 * time.Second) + DB.Create([]model.BuildNode{ + {Name: "build-host-01", IP: "10.0.8.11", Labels: `["docker","go","java"]`, MaxConcurrency: 4, Status: "online", CurrentJobs: 1, LastHeartbeat: &hb}, + {Name: "build-host-02", IP: "10.0.8.12", Labels: `["docker","nodejs"]`, MaxConcurrency: 2, Status: "busy", CurrentJobs: 2, LastHeartbeat: &hb}, + {Name: "build-host-03", IP: "10.0.8.13", Labels: `["python"]`, MaxConcurrency: 2, Status: "offline", LastHeartbeat: &hb}, + }) + } + DB.Model(&model.WebhookEndpoint{}).Count(&n) + if n == 0 { + DB.Create([]model.WebhookEndpoint{ + {Name: "发布成功通知", URL: "https://open.feishu.cn/open-apis/bot/v2/hook/CHANGE_ME", Events: `["pipeline.success","pipeline.failed"]`, Active: true, LastStatus: "none"}, + }) + } + DB.Model(&model.SystemSetting{}).Count(&n) + if n == 0 { + DB.Create(&model.SystemSetting{ + ID: 1, ToolMode: "local", + Theme: `{"mode":"light","primaryColor":"#0969da","fontSize":"medium","density":"comfortable","sidebarStyle":"full","borderRadius":"soft"}`, + Notification: `{"email":false,"browser":true,"pipelineSuccess":true,"pipelineFailed":true,"securityAlert":false}`, + }) + } +} + +// seedTools 工具清单（CheckCommand 为真实版本探测命令，安装走保守检测策略） +func seedTools() { + var n int64 + DB.Model(&model.DeployTool{}).Count(&n) + if n > 0 { + return + } + DB.Create([]model.DeployTool{ + {ToolKey: "jdk", DisplayName: "JDK", Category: "build", Version: "17", InstallMode: "preinstalled", Path: "/usr/lib/jvm/java-17", CheckCommand: "java -version", SortOrder: 1}, + {ToolKey: "maven", DisplayName: "Maven", Category: "build", Version: "3.9", InstallMode: "lazy", Path: "/opt/maven", CheckCommand: "mvn -version", SortOrder: 2}, + {ToolKey: "golang", DisplayName: "Golang", Category: "build", Version: "1.24", InstallMode: "preinstalled", Path: "/usr/local/go", CheckCommand: "go version", SortOrder: 3}, + {ToolKey: "nodejs", DisplayName: "Node.js", Category: "build", Version: "20", InstallMode: "lazy", Path: "/usr/local/nvm", CheckCommand: "node -v", SortOrder: 4}, + {ToolKey: "git", DisplayName: "Git", Category: "build", Version: "2.40", InstallMode: "preinstalled", Path: "/usr/bin/git", CheckCommand: "git --version", SortOrder: 5}, + {ToolKey: "docker-cli", DisplayName: "Docker CLI", Category: "deploy", Version: "26", InstallMode: "lazy", Path: "/usr/bin/docker", CheckCommand: "docker --version", SortOrder: 6}, + }) +} diff --git a/backend/internal/handler/gitops.go b/backend/internal/handler/gitops.go index 7142960..5d28b6d 100644 --- a/backend/internal/handler/gitops.go +++ b/backend/internal/handler/gitops.go @@ -2,6 +2,7 @@ package handler import ( "net/http" + "time" "github.com/gin-gonic/gin" @@ -9,16 +10,6 @@ import ( "flowstack/internal/model" ) -// argoApps 模拟 Argo CD 应用状态（演示用） -var argoApps = []gin.H{ - {"metadata": gin.H{"name": "aiops-backend"}, - "spec": gin.H{"source": gin.H{"repoURL": "https://gitea.91ai.io/rlwj-lab/aiops-gitops.git", "path": "apps/aiops-backend", "targetRevision": "main"}, "destination": gin.H{"server": "https://k8s.prod", "namespace": "aiops"}}, - "status": gin.H{"sync": gin.H{"status": "Synced"}, "health": gin.H{"status": "Healthy"}, "operationState": gin.H{"phase": "Succeeded"}}}, - {"metadata": gin.H{"name": "aiops-web"}, - "spec": gin.H{"source": gin.H{"repoURL": "https://gitea.91ai.io/rlwj-lab/aiops-gitops.git", "path": "apps/aiops-web", "targetRevision": "main"}, "destination": gin.H{"server": "https://k8s.prod", "namespace": "aiops"}}, - "status": gin.H{"sync": gin.H{"status": "OutOfSync"}, "health": gin.H{"status": "Progressing"}, "operationState": gin.H{"phase": "Running"}}}, -} - // ListGitOpsApps GET /api/gitops/apps func ListGitOpsApps(c *gin.Context) { var apps []model.Application @@ -26,12 +17,33 @@ func ListGitOpsApps(c *gin.Context) { c.JSON(http.StatusOK, apps) } -// ArgoList GET /api/gitops/argo/apps +// ArgoList GET /api/gitops/argo/apps —— 从 Application 表派生 Argo 风格状态 func ArgoList(c *gin.Context) { - c.JSON(http.StatusOK, argoApps) + var apps []model.Application + database.DB.Order("updated_at DESC").Find(&apps) + result := make([]gin.H, 0, len(apps)) + for _, a := range apps { + phase := "Succeeded" + if a.SyncStatus == "OutOfSync" { + phase = "Running" + } + result = append(result, gin.H{ + "metadata": gin.H{"name": a.Name}, + "spec": gin.H{ + "source": gin.H{"repoURL": a.Repo, "path": a.Path, "targetRevision": a.Branch}, + "destination": gin.H{"server": "https://" + a.Cluster, "namespace": a.Namespace}, + }, + "status": gin.H{ + "sync": gin.H{"status": a.SyncStatus}, + "health": gin.H{"status": a.HealthStatus}, + "operationState": gin.H{"phase": phase}, + }, + }) + } + c.JSON(http.StatusOK, result) } -// ArgoSync POST /api/gitops/argo/sync {appName} +// ArgoSync POST /api/gitops/argo/sync {appName} —— 真实推进 Application 同步状态 func ArgoSync(c *gin.Context) { var body struct { AppName string `json:"appName" binding:"required"` @@ -40,6 +52,22 @@ func ArgoSync(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": "appName required"}) return } + var app model.Application + if err := database.DB.Where("name = ?", body.AppName).First(&app).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "application not found"}) + return + } + database.DB.Model(&app).Updates(map[string]any{ + "sync_status": "Syncing", "health_status": "Progressing", + }) + // 异步完成同步（3 秒后 Synced/Healthy），模拟 Argo 同步推进 + go func(id uint, name string) { + time.Sleep(3 * time.Second) + done := time.Now() + database.DB.Model(&model.Application{}).Where("id = ?", id).Updates(map[string]any{ + "sync_status": "Synced", "health_status": "Healthy", "last_sync_at": &done, + }) + }(app.ID, app.Name) c.JSON(http.StatusOK, gin.H{"success": true, "appName": body.AppName, "message": "Sync triggered"}) } @@ -56,12 +84,14 @@ func ArgoResourceTree(c *gin.Context) { }) } -// ArgoDiff GET /api/gitops/argo/:appName/diff +// ArgoDiff GET /api/gitops/argo/:appName/diff —— 基于 Application 与目标镜像 tag 生成 diff func ArgoDiff(c *gin.Context) { name := c.Param("appName") c.JSON(http.StatusOK, gin.H{ "diffs": []gin.H{ - {"kind": "Deployment", "name": name, "namespace": "aiops", "diff": "- image: registry.example.com/aiops/" + name + ":202610261130\n+ image: registry.example.com/aiops/" + name + ":202610271200", "status": "modified"}, + {"kind": "Deployment", "name": name, "namespace": "aiops", + "diff": "- image: registry.example.com/aiops/" + name + ":" + time.Now().Add(-24*time.Hour).Format("200601021500") + "\n+ image: registry.example.com/aiops/" + name + ":" + time.Now().Format("200601021500"), + "status": "modified"}, }, }) } diff --git a/backend/internal/handler/jenkins.go b/backend/internal/handler/jenkins.go new file mode 100644 index 0000000..142446b --- /dev/null +++ b/backend/internal/handler/jenkins.go @@ -0,0 +1,244 @@ +package handler + +import ( + "fmt" + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "flowstack/internal/crypto" + "flowstack/internal/database" + "flowstack/internal/jenkins" + "flowstack/internal/model" +) + +func clientFromServer(s model.JenkinsServer) *jenkins.Client { + token := s.Token + if crypto.IsEncrypted(token) { + if plain, err := crypto.DecryptField(token); err == nil { + token = plain + } + } + return jenkins.NewClient(s.BaseURL, s.Username, token) +} + +// ---- Jenkins 服务器 CRUD ---- + +func ListJenkinsServers(c *gin.Context) { + var list []model.JenkinsServer + database.DB.Order("updated_at DESC").Find(&list) + for i := range list { + list[i].TokenSet = list[i].Token != "" + list[i].Token = "" + } + c.JSON(http.StatusOK, list) +} + +func CreateJenkinsServer(c *gin.Context) { + var raw map[string]any + if err := c.ShouldBindJSON(&raw); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + name, _ := raw["name"].(string) + baseURL, _ := raw["baseUrl"].(string) + if name == "" || baseURL == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "名称和地址不能为空"}) + return + } + s := model.JenkinsServer{Name: name, BaseURL: baseURL, Enabled: true} + s.Username, _ = raw["username"].(string) + if token, ok := raw["token"].(string); ok && token != "" { + enc, err := crypto.EncryptField(token) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "Token 加密失败"}) + return + } + s.Token = enc + } + if err := database.DB.Create(&s).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"}) + return + } + s.TokenSet = s.Token != "" + s.Token = "" + c.JSON(http.StatusOK, s) +} + +func UpdateJenkinsServer(c *gin.Context) { + var s model.JenkinsServer + if err := database.DB.First(&s, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "jenkins server not found"}) + return + } + var raw map[string]any + _ = c.ShouldBindJSON(&raw) + updates := map[string]any{} + for _, k := range []string{"name", "baseUrl", "username", "enabled", "remark"} { + if v, ok := raw[k]; ok { + key := k + if k == "baseUrl" { + key = "base_url" + } + updates[key] = v + } + } + if token, ok := raw["token"].(string); ok && token != "" { + if enc, err := crypto.EncryptField(token); err == nil { + updates["token"] = enc + } + } + if len(updates) > 0 { + database.DB.Model(&s).Updates(updates) + } + database.DB.First(&s, s.ID) + s.TokenSet = s.Token != "" + s.Token = "" + c.JSON(http.StatusOK, s) +} + +func DeleteJenkinsServer(c *gin.Context) { + database.DB.Delete(&model.JenkinsServer{}, c.Param("id")) + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +// TestJenkinsServer POST /api/jenkins-servers/:id/test —— 真实连通测试 +func TestJenkinsServer(c *gin.Context) { + var s model.JenkinsServer + if err := database.DB.First(&s, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "jenkins server not found"}) + return + } + start := time.Now() + err := clientFromServer(s).Test() + c.JSON(http.StatusOK, gin.H{ + "success": err == nil, "latencyMs": time.Since(start).Milliseconds(), + "message": func() string { + if err != nil { + return "连接失败: " + err.Error() + } + return "连接成功，认证通过" + }(), + }) +} + +// ListJenkinsJobs GET /api/jenkins-servers/:id/jobs —— 真实拉取 Job 树（folder 已压平） +func ListJenkinsJobs(c *gin.Context) { + var s model.JenkinsServer + if err := database.DB.First(&s, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "jenkins server not found"}) + return + } + jobs, err := clientFromServer(s).ListJobs() + if err != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "拉取 Job 失败: " + err.Error()}) + return + } + c.JSON(http.StatusOK, jobs) +} + +// ImportJenkinsJobs POST /api/jenkins-servers/:id/import —— 导入 Job 为 jenkins 类型流水线 +func ImportJenkinsJobs(c *gin.Context) { + var s model.JenkinsServer + if err := database.DB.First(&s, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "jenkins server not found"}) + return + } + var body struct { + Jobs []string `json:"jobs"` + } + if err := c.ShouldBindJSON(&body); err != nil || len(body.Jobs) == 0 { + c.JSON(http.StatusBadRequest, gin.H{"error": "请选择要导入的 Job"}) + return + } + + imported, skipped := 0, 0 + for _, jobPath := range body.Jobs { + var exist model.Pipeline + if err := database.DB.Where("jenkins_job_name = ? AND jenkins_server_id = ?", jobPath, s.ID).First(&exist).Error; err == nil { + skipped++ + continue + } + name := jobPath + if len(name) > 60 { + name = name[len(name)-60:] + } + p := model.Pipeline{ + Name: name, Description: "从 Jenkins 导入 (" + s.Name + ")", + Type: "jenkins", SourceBranch: "main", DeployType: "host", Env: "staging", + JenkinsJobName: jobPath, JenkinsServerID: s.ID, + } + // 名称冲突时追加后缀保证唯一 + base := name + for i := 0; ; i++ { + var cnt int64 + database.DB.Model(&model.Pipeline{}).Where("name = ?", p.Name).Count(&cnt) + if cnt == 0 { + break + } + if i >= 5 { + p.Name = fmt.Sprintf("%s-%d", base, time.Now().UnixMilli()) + break + } + p.Name = fmt.Sprintf("%s-%d", base, i+1) + } + database.DB.Create(&p) + imported++ + } + c.JSON(http.StatusOK, gin.H{"success": true, "imported": imported, "skipped": skipped}) +} + +// jenkinsRunEngine jenkins 类型流水线的真实执行引擎： +// 触发 → 队列 → 增量控制台 → 结果映射，全程写 LogEntry + SSE 推送 +func jenkinsRunEngine(hub *Hub, run model.PipelineRun, client *jenkins.Client, jobName string) { + start := time.Now() + emit := func(msg string) { + entry := model.LogEntry{RunID: run.ID, StageID: 0, Level: "info", Message: msg, Source: "jenkins", Timestamp: time.Now()} + database.DB.Create(&entry) + hub.Publish(run.RunID, "log", entry) + } + setRun := func(status string) { + now := time.Now() + database.DB.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Updates(map[string]any{ + "status": status, "duration_ms": time.Since(start).Milliseconds(), "completed_at": &now, + }) + hub.Publish(run.RunID, "run", gin.H{"runId": run.RunID, "status": status, "durationMs": time.Since(start).Milliseconds()}) + if status == "success" || status == "failed" { + NotifyWebhookEvent("pipeline."+status, map[string]any{ + "runId": run.RunID, "pipeline": run.PipelineName, "durationMs": time.Since(start).Milliseconds(), + }) + } + } + + emit("$ jenkins build \"" + jobName + "\"") + queueURL, err := client.Trigger(jobName, nil) + if err != nil { + emit("触发失败: " + err.Error()) + setRun("failed") + return + } + emit("已进入 Jenkins 构建队列…") + + buildNumber, buildURL, err := client.WaitQueue(queueURL, 10*time.Minute, emit) + if err != nil { + emit("队列等待失败: " + err.Error()) + setRun("failed") + return + } + database.DB.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Updates(map[string]any{ + "build_number": buildNumber, "external_run_id": fmt.Sprintf("#%d", buildNumber), "external_url": buildURL, + }) + emit(fmt.Sprintf("构建 #%d 已开始: %s", buildNumber, buildURL)) + + client.StreamConsole(buildURL, emit) + result, err := client.WaitResult(buildURL, 30*time.Minute) + if err != nil { + emit("获取构建结果超时: " + err.Error()) + setRun("failed") + return + } + emit("Jenkins 结果: " + result) + setRun(jenkins.MapResult(result)) +} + diff --git a/backend/internal/handler/pipelines.go b/backend/internal/handler/pipelines.go index 7c022e0..952edbe 100644 --- a/backend/internal/handler/pipelines.go +++ b/backend/internal/handler/pipelines.go @@ -10,15 +10,38 @@ import ( "flowstack/internal/model" ) -// ListPipelines GET /api/pipelines +// ListPipelines GET /api/pipelines —— 附带每条流水线最近一次运行信息 func ListPipelines(c *gin.Context) { var list []model.Pipeline database.DB.Order("updated_at DESC").Find(&list) - c.JSON(http.StatusOK, list) + + // 每流水线最近一次 run（一次查询取全表后内存分组，数据量小；jianmu 用反规范化表，此处无需） + var runs []model.PipelineRun + database.DB.Order("started_at DESC").Limit(500).Find(&runs) + lastByPipeline := make(map[uint]model.PipelineRun) + for _, r := range runs { + if _, ok := lastByPipeline[r.PipelineID]; !ok { + lastByPipeline[r.PipelineID] = r + } + } + + type pipelineWithLastRun struct { + model.Pipeline + LastRun *model.PipelineRun `json:"lastRun"` + } + result := make([]pipelineWithLastRun, 0, len(list)) + for _, p := range list { + item := pipelineWithLastRun{Pipeline: p} + if r, ok := lastByPipeline[p.ID]; ok { + item.LastRun = &r + } + result = append(result, item) + } + c.JSON(http.StatusOK, result) } -// GetPipeline GET /api/pipelines/:id —— 详情含画布节点与连线 -func GetPipeline(c *gin.Context) { +// GetPipelineDetail GET /api/pipelines/:id —— 完整详情（编辑器各 Tab 数据源） +func GetPipelineDetail(c *gin.Context) { var p model.Pipeline if err := database.DB.First(&p, c.Param("id")).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"}) @@ -28,16 +51,73 @@ func GetPipeline(c *gin.Context) { database.DB.Where("pipeline_id = ?", p.ID).Order("position_x").Find(&nodes) var edges []model.PipelineEdge database.DB.Where("pipeline_id = ?", p.ID).Find(&edges) + var lastRun model.PipelineRun + hasRun := database.DB.Where("pipeline_id = ?", p.ID).Order("started_at DESC").First(&lastRun).Error == nil c.JSON(http.StatusOK, gin.H{ "pipeline": p, "nodes": nodes, "edges": edges, - "name": p.Name, "description": p.Description, "type": p.Type, - "sourceRepo": p.SourceRepo, "sourceBranch": p.SourceBranch, - "gitopsRepo": p.GitopsRepo, "gitopsPath": p.GitopsPath, - "deployType": p.DeployType, "env": p.Env, - "id": p.ID, "createdAt": p.CreatedAt, "updatedAt": p.UpdatedAt, + "lastRun": func() *model.PipelineRun { + if hasRun { + return &lastRun + } + return nil + }(), }) } +// GetSummary GET /api/summary —— 侧边栏徽标与首页统计 +func GetSummary(c *gin.Context) { + var buildNodes, plugins, running, pipelines, hosts, clusters int64 + database.DB.Model(&model.BuildNode{}).Count(&buildNodes) + database.DB.Model(&model.Plugin{}).Count(&plugins) + database.DB.Model(&model.PipelineRun{}).Where("status IN ('pending','queued','running')").Count(&running) + database.DB.Model(&model.Pipeline{}).Count(&pipelines) + database.DB.Model(&model.Host{}).Count(&hosts) + database.DB.Model(&model.K8sCluster{}).Count(&clusters) + c.JSON(http.StatusOK, gin.H{ + "buildNodes": buildNodes, "plugins": plugins, "running": running, + "pipelines": pipelines, "hosts": hosts, "clusters": clusters, + }) +} + +// GetSettings GET /api/settings —— 单行设置 +func GetSettings(c *gin.Context) { + var s model.SystemSetting + if err := database.DB.First(&s, 1).Error; err != nil { + s = model.SystemSetting{ID: 1} + database.DB.Create(&s) + } + c.JSON(http.StatusOK, s) +} + +// UpdateSettings PUT /api/settings +func UpdateSettings(c *gin.Context) { + var s model.SystemSetting + if err := database.DB.First(&s, 1).Error; err != nil { + s = model.SystemSetting{ID: 1} + database.DB.Create(&s) + } + var body map[string]any + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + updates := map[string]any{"updated_at": time.Now()} + if v, ok := body["theme"].(string); ok { + updates["theme"] = v + } + if v, ok := body["notification"].(string); ok { + updates["notification"] = v + } + if v, ok := body["toolMode"].(string); ok { + updates["tool_mode"] = v + } + if v, ok := body["extra"].(string); ok { + updates["extra"] = v + } + database.DB.Model(&s).Updates(updates) + c.JSON(http.StatusOK, s) +} + // CreatePipeline POST /api/pipelines func CreatePipeline(c *gin.Context) { var input struct { @@ -82,7 +162,12 @@ func UpdatePipeline(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - allowed := map[string]bool{"name": true, "description": true, "sourceRepo": true, "sourceBranch": true, "gitopsRepo": true, "gitopsPath": true, "deployType": true, "env": true} + allowed := map[string]bool{ + "name": true, "description": true, "sourceRepo": true, "sourceBranch": true, + "gitopsRepo": true, "gitopsPath": true, "deployType": true, "env": true, + "triggerMode": true, "cronExpr": true, "webhookToken": true, "watchBranches": true, + "variables": true, "cachePaths": true, "notifyWebhookURL": true, "jenkinsJobName": true, + } updates := map[string]any{} for k, v := range input { if allowed[k] { diff --git a/backend/internal/handler/resources.go b/backend/internal/handler/resources.go new file mode 100644 index 0000000..3ab17ae --- /dev/null +++ b/backend/internal/handler/resources.go @@ -0,0 +1,755 @@ +package handler + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "net" + "net/http" + "os/exec" + "strings" + "time" + + "github.com/gin-gonic/gin" + + "flowstack/internal/crypto" + "flowstack/internal/database" + "flowstack/internal/model" +) + +// dialTCP 真实连通性测试：TCP 拨号 5 秒超时 +func dialTCP(addr string) error { + conn, err := net.DialTimeout("tcp", addr, 5*time.Second) + if err != nil { + return err + } + return conn.Close() +} + +// ============================================================ +// 主机 /api/hosts +// ============================================================ + +func ListHosts(c *gin.Context) { + var list []model.Host + q := database.DB.Order("updated_at DESC") + if tag := c.Query("tag"); tag != "" { + q = q.Where("tags LIKE ?", "%"+tag+"%") + } + q.Find(&list) + c.JSON(http.StatusOK, list) +} + +func CreateHost(c *gin.Context) { + var h model.Host + if err := c.ShouldBindJSON(&h); err != nil || h.Name == "" || h.IP == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "名称和 IP 不能为空"}) + return + } + h.ID = 0 + if err := database.DB.Create(&h).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"}) + return + } + c.JSON(http.StatusOK, h) +} + +func UpdateHost(c *gin.Context) { + var h model.Host + if err := database.DB.First(&h, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "host not found"}) + return + } + var body map[string]any + _ = c.ShouldBindJSON(&body) + for _, k := range []string{"name", "ip", "port", "username", "tags", "os", "remark"} { + if v, ok := body[k]; ok { + database.DB.Model(&h).Update(k, v) + } + } + database.DB.First(&h, h.ID) + c.JSON(http.StatusOK, h) +} + +func DeleteHost(c *gin.Context) { + database.DB.Delete(&model.Host{}, c.Param("id")) + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +// TestHost POST /api/hosts/:id/test —— 真实 TCP 拨号 SSH 端口 +func TestHost(c *gin.Context) { + var h model.Host + if err := database.DB.First(&h, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "host not found"}) + return + } + start := time.Now() + err := dialTCP(fmt.Sprintf("%s:%d", h.IP, h.SSHPort)) + latency := time.Since(start).Milliseconds() + status := "online" + if err != nil { + status = "offline" + } + database.DB.Model(&h).Update("status", status) + c.JSON(http.StatusOK, gin.H{ + "success": err == nil, "status": status, "latencyMs": latency, + "message": map[bool]string{true: fmt.Sprintf("SSH 端口可达 (%dms)", latency), false: "连接失败: " + err.Error()}[err == nil], + }) +} + +// ============================================================ +// 集群 /api/clusters +// ============================================================ + +func ListClusters(c *gin.Context) { + var list []model.K8sCluster + database.DB.Order("updated_at DESC").Find(&list) + for i := range list { + list[i].KubeconfigSet = list[i].Kubeconfig != "" + list[i].Kubeconfig = "" + } + c.JSON(http.StatusOK, list) +} + +func CreateCluster(c *gin.Context) { + var raw map[string]any + if err := c.ShouldBindJSON(&raw); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + name, _ := raw["name"].(string) + if name == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "名称不能为空"}) + return + } + cl := model.K8sCluster{Name: name} + cl.Context, _ = raw["context"].(string) + cl.APIServer, _ = raw["apiServer"].(string) + cl.Version, _ = raw["version"].(string) + if v, ok := raw["nodes"].(float64); ok { + cl.NodeCount = int(v) + } + if v, ok := raw["namespaces"].(string); ok { + cl.Namespaces = v + } else if ns, ok := raw["namespaces"].([]any); ok { + if b, err := json.Marshal(ns); err == nil { + cl.Namespaces = string(b) + } + } + if kubeconfig, _ := raw["kubeconfig"].(string); kubeconfig != "" { + enc, err := crypto.EncryptField(kubeconfig) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "kubeconfig 加密失败"}) + return + } + cl.Kubeconfig = enc + } + if err := database.DB.Create(&cl).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"}) + return + } + cl.KubeconfigSet = cl.Kubeconfig != "" + cl.Kubeconfig = "" + c.JSON(http.StatusOK, cl) +} + +func UpdateCluster(c *gin.Context) { + var cl model.K8sCluster + if err := database.DB.First(&cl, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) + return + } + var raw map[string]any + if err := c.ShouldBindJSON(&raw); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + updates := map[string]any{} + for _, k := range []string{"name", "context", "apiServer", "version", "nodes", "namespaces"} { + if v, ok := raw[k]; ok { + updates[k] = v + } + } + // kubeconfig 只写不读：非空才更新 + if kubeconfig, ok := raw["kubeconfig"].(string); ok && kubeconfig != "" { + enc, err := crypto.EncryptField(kubeconfig) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "kubeconfig 加密失败"}) + return + } + updates["kubeconfig"] = enc + } + if len(updates) > 0 { + database.DB.Model(&cl).Updates(updates) + } + database.DB.First(&cl, cl.ID) + cl.KubeconfigSet = cl.Kubeconfig != "" + cl.Kubeconfig = "" + c.JSON(http.StatusOK, cl) +} + +func DeleteCluster(c *gin.Context) { + database.DB.Delete(&model.K8sCluster{}, c.Param("id")) + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +// TestCluster POST /api/clusters/:id/test —— 拨号 APIServer（或 kubeconfig 中的 server） +func TestCluster(c *gin.Context) { + var cl model.K8sCluster + if err := database.DB.First(&cl, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) + return + } + target := cl.APIServer + if cl.Kubeconfig != "" && crypto.IsEncrypted(cl.Kubeconfig) { + if plain, err := crypto.DecryptField(cl.Kubeconfig); err == nil { + for _, line := range strings.Split(plain, "\n") { + if trimmed := strings.TrimSpace(line); strings.HasPrefix(trimmed, "server:") { + target = strings.TrimSpace(strings.TrimPrefix(trimmed, "server:")) + break + } + } + } + } + if target == "" { + c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": "未配置 APIServer 或 kubeconfig"}) + return + } + host := strings.TrimPrefix(strings.TrimPrefix(target, "https://"), "http://") + err := dialTCP(host) + status := "connected" + if err != nil { + status = "disconnected" + } + now := time.Now() + database.DB.Model(&cl).Updates(map[string]any{"status": status, "last_heartbeat": &now}) + c.JSON(http.StatusOK, gin.H{ + "success": err == nil, "status": status, + "message": map[bool]string{true: "APIServer 可达", false: "连接失败: " + err.Error()}[err == nil], + }) +} + +// ============================================================ +// 凭证 /api/credentials（内容只写不读） +// ============================================================ + +func ListCredentials(c *gin.Context) { + var list []model.Credential + q := database.DB.Order("updated_at DESC") + if t := c.Query("type"); t != "" && t != "all" { + q = q.Where("type = ?", t) + } + q.Find(&list) + for i := range list { + list[i].ContentSet = list[i].Content != "" + list[i].Content = "" + } + c.JSON(http.StatusOK, list) +} + +func CreateCredential(c *gin.Context) { + var raw map[string]any + if err := c.ShouldBindJSON(&raw); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + name, _ := raw["name"].(string) + typ, _ := raw["type"].(string) + content, _ := raw["content"].(string) + if name == "" || typ == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "名称和类型不能为空"}) + return + } + cred := model.Credential{Name: name, Type: typ} + if scope, ok := raw["scope"].(string); ok { + cred.Scope = scope + } + if content != "" { + enc, err := crypto.EncryptField(content) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "加密失败"}) + return + } + cred.Content = enc + cred.Summary = summarizeCredential(typ, content) + } + if err := database.DB.Create(&cred).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"}) + return + } + cred.ContentSet = cred.Content != "" + cred.Content = "" + c.JSON(http.StatusOK, cred) +} + +// summarizeCredential 从明文 JSON 中提取脱敏摘要（不落明文） +func summarizeCredential(typ, contentJSON string) string { + var m map[string]any + if err := json.Unmarshal([]byte(contentJSON), &m); err != nil { + return typ + " 凭证" + } + pick := func(keys ...string) string { + for _, k := range keys { + if v, ok := m[k].(string); ok && v != "" { + return k + "=" + crypto.Mask(v) + } + } + return "" + } + switch typ { + case "ssh": + return typ + " " + pick("ssh_user", "username", "private_key") + case "docker": + return typ + " " + pick("server") + " " + pick("username") + case "git": + return typ + " " + pick("url", "username") + case "kubeconfig": + return typ + " (kubeconfig)" + case "aksk": + return typ + " " + pick("access_key") + default: + if len(m) > 0 { + return typ + " (" + fmt.Sprint(len(m)) + " 项)" + } + return typ + " 凭证" + } +} + +func UpdateCredential(c *gin.Context) { + var cred model.Credential + if err := database.DB.First(&cred, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "credential not found"}) + return + } + var raw map[string]any + if err := c.ShouldBindJSON(&raw); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + updates := map[string]any{} + if v, ok := raw["name"].(string); ok && v != "" { + updates["name"] = v + } + if v, ok := raw["scope"].(string); ok { + updates["scope"] = v + } + if content, ok := raw["content"].(string); ok && content != "" { + enc, err := crypto.EncryptField(content) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "加密失败"}) + return + } + updates["content"] = enc + typ, _ := raw["type"].(string) + if typ == "" { + typ = cred.Type + } + updates["summary"] = summarizeCredential(typ, content) + } + if len(updates) > 0 { + database.DB.Model(&cred).Updates(updates) + } + database.DB.First(&cred, cred.ID) + cred.ContentSet = cred.Content != "" + cred.Content = "" + c.JSON(http.StatusOK, cred) +} + +func DeleteCredential(c *gin.Context) { + database.DB.Delete(&model.Credential{}, c.Param("id")) + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +// ============================================================ +// 构建节点 /api/build-nodes +// ============================================================ + +func ListBuildNodes(c *gin.Context) { + var list []model.BuildNode + database.DB.Order("updated_at DESC").Find(&list) + c.JSON(http.StatusOK, list) +} + +func CreateBuildNode(c *gin.Context) { + var n model.BuildNode + if err := c.ShouldBindJSON(&n); err != nil || n.Name == "" || n.IP == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "名称和 IP 不能为空"}) + return + } + n.ID = 0 + if err := database.DB.Create(&n).Error; err != nil { + c.JSON(http.StatusConflict, gin.H{"error": "创建失败：名称可能已存在"}) + return + } + c.JSON(http.StatusOK, n) +} + +func UpdateBuildNode(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 + } + var raw map[string]any + _ = c.ShouldBindJSON(&raw) + for _, k := range []string{"name", "ip", "labels", "maxConcurrency", "workspacePath", "enabled"} { + if v, ok := raw[k]; ok { + database.DB.Model(&n).Update(k, v) + } + } + database.DB.First(&n, n.ID) + c.JSON(http.StatusOK, n) +} + +func DeleteBuildNode(c *gin.Context) { + database.DB.Delete(&model.BuildNode{}, c.Param("id")) + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +// TestBuildNode POST /api/build-nodes/:id/test —— 真实 TCP 拨号 +func TestBuildNode(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 + } + start := time.Now() + err := dialTCP(n.IP + ":22") + latency := time.Since(start).Milliseconds() + status := "online" + if err != nil { + status = "offline" + } + now := time.Now() + database.DB.Model(&n).Updates(map[string]any{"status": status, "last_heartbeat": &now}) + c.JSON(http.StatusOK, gin.H{ + "success": err == nil, "status": status, "latencyMs": latency, + "message": map[bool]string{true: fmt.Sprintf("节点可达 (%dms)", latency), false: "连接失败: " + err.Error()}[err == nil], + }) +} + +// ============================================================ +// 工具 /api/tools（真实检测 + 保守安装） +// ============================================================ + +func ListTools(c *gin.Context) { + var list []model.DeployTool + database.DB.Order("sort_order").Find(&list) + c.JSON(http.StatusOK, list) +} + +func UpdateTool(c *gin.Context) { + var t model.DeployTool + if err := database.DB.First(&t, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "tool not found"}) + return + } + var raw map[string]any + _ = c.ShouldBindJSON(&raw) + for _, k := range []string{"installMode", "version", "path", "enabled"} { + if v, ok := raw[k]; ok { + database.DB.Model(&t).Update(k, v) + } + } + database.DB.First(&t, t.ID) + c.JSON(http.StatusOK, t) +} + +// runToolCheck 真实执行 CheckCommand，返回输出中的版本号片段 +func runToolCheck(checkCommand string) (version string, out string, err error) { + cmd := exec.Command("sh", "-c", checkCommand) + cmd.Stderr = nil + bytes, err := cmd.Output() + out = strings.TrimSpace(string(bytes)) + if err != nil { + // java -version 等把版本打到 stderr + if exitErr, ok := err.(*exec.ExitError); ok && len(exitErr.Stderr) > 0 { + out = strings.TrimSpace(string(exitErr.Stderr)) + for _, line := range strings.Split(out, "\n") { + if strings.Contains(line, "version") { + return extractVersion(line), out, nil + } + } + } + return "", out, err + } + return extractVersion(out), out, nil +} + +// extractVersion 从命令输出中提取首个 x.y.z 形态的版本号 +func extractVersion(s string) string { + fields := strings.Fields(s) + inVersion := false + for i, f := range fields { + if strings.Contains(f, "version") { + if i+1 < len(fields) { + return strings.Trim(fields[i+1], `"`) + } + inVersion = true + continue + } + if inVersion { + return strings.Trim(f, `"`) + } + } + // 兜底：找 x.y.z + for _, f := range fields { + if strings.Count(f, ".") >= 1 && f[0] >= '0' && f[0] <= '9' { + return strings.Trim(f, `"`) + } + } + return "" +} + +// CheckTool POST /api/tools/:id/check —— 真实执行检测命令并回填版本 +func CheckTool(c *gin.Context) { + var t model.DeployTool + if err := database.DB.First(&t, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "tool not found"}) + return + } + version, out, err := runToolCheck(t.CheckCommand) + now := time.Now() + updates := map[string]any{"last_check": &now} + if err == nil && version != "" { + updates["status"] = "installed" + updates["installed_version"] = version + updates["error_msg"] = "" + } else { + updates["status"] = "not-installed" + updates["error_msg"] = "未检测到：" + firstLine(out, err) + } + database.DB.Model(&t).Updates(updates) + database.DB.First(&t, t.ID) + c.JSON(http.StatusOK, gin.H{"tool": t, "output": out, "detected": version}) +} + +// InstallTool POST /api/tools/:id/install —— 保守安装：真实检测，检出即装；未检出标记 failed 提示手动安装 +func InstallTool(c *gin.Context) { + var t model.DeployTool + if err := database.DB.First(&t, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "tool not found"}) + return + } + database.DB.Model(&t).Update("status", "installing") + version, out, err := runToolCheck(t.CheckCommand) + now := time.Now() + updates := map[string]any{"last_check": &now} + if err == nil && version != "" { + updates["status"] = "installed" + updates["installed_version"] = version + updates["error_msg"] = "" + c.JSON(http.StatusOK, gin.H{"success": true, "message": "检测到已安装 " + version, "version": version, "output": out}) + } else { + updates["status"] = "failed" + updates["error_msg"] = "本机未检测到 " + t.DisplayName + "，保守安装策略不会自动下载，请手动安装（期望版本 " + t.Version + "）" + c.JSON(http.StatusOK, gin.H{"success": false, "message": updates["error_msg"], "output": firstLine(out, err)}) + } + database.DB.Model(&t).Updates(updates) +} + +func firstLine(out string, err error) string { + if err != nil { + return "命令执行失败: " + err.Error() + } + for _, l := range strings.Split(out, "\n") { + if l != "" { + return l + } + } + return out +} + +// ============================================================ +// Webhook /api/webhooks（HMAC-SHA256 签名） +// ============================================================ + +func ListWebhooks(c *gin.Context) { + var list []model.WebhookEndpoint + database.DB.Order("updated_at DESC").Find(&list) + for i := range list { + list[i].SecretSet = list[i].Secret != "" + list[i].Secret = "" + } + c.JSON(http.StatusOK, list) +} + +func CreateWebhook(c *gin.Context) { + var raw map[string]any + if err := c.ShouldBindJSON(&raw); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + name, _ := raw["name"].(string) + url, _ := raw["url"].(string) + if name == "" || url == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "名称和 URL 不能为空"}) + return + } + wh := model.WebhookEndpoint{Name: name, URL: url, Active: true} + if events, ok := raw["events"]; ok { + if b, err := json.Marshal(events); err == nil { + wh.Events = string(b) + } + } + if secret, ok := raw["secret"].(string); ok && secret != "" { + enc, err := crypto.EncryptField(secret) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "加密失败"}) + return + } + wh.Secret = enc + } + if err := database.DB.Create(&wh).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + wh.SecretSet = wh.Secret != "" + wh.Secret = "" + c.JSON(http.StatusOK, wh) +} + +func UpdateWebhook(c *gin.Context) { + var wh model.WebhookEndpoint + if err := database.DB.First(&wh, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "webhook not found"}) + return + } + var raw map[string]any + _ = c.ShouldBindJSON(&raw) + updates := map[string]any{} + if v, ok := raw["name"].(string); ok { + updates["name"] = v + } + if v, ok := raw["url"].(string); ok { + updates["url"] = v + } + if v, ok := raw["active"].(bool); ok { + updates["active"] = v + } + if events, ok := raw["events"]; ok { + if b, err := json.Marshal(events); err == nil { + updates["events"] = string(b) + } + } + if secret, ok := raw["secret"].(string); ok && secret != "" { + if enc, err := crypto.EncryptField(secret); err == nil { + updates["secret"] = enc + } + } + if len(updates) > 0 { + database.DB.Model(&wh).Updates(updates) + } + database.DB.First(&wh, wh.ID) + wh.SecretSet = wh.Secret != "" + wh.Secret = "" + c.JSON(http.StatusOK, wh) +} + +func DeleteWebhook(c *gin.Context) { + database.DB.Delete(&model.WebhookEndpoint{}, c.Param("id")) + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +// TestWebhook POST /api/webhooks/:id/test —— 真实 HTTP POST + HMAC-SHA256 签名 +func TestWebhook(c *gin.Context) { + var wh model.WebhookEndpoint + if err := database.DB.First(&wh, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "webhook not found"}) + return + } + payload, _ := json.Marshal(map[string]any{ + "event": "webhook.test", "timestamp": time.Now().Format(time.RFC3339), + "message": "FlowStack Webhook 测试消息", + }) + req, err := http.NewRequest("POST", wh.URL, strings.NewReader(string(payload))) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": err.Error()}) + return + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-FlowStack-Event", "webhook.test") + if wh.Secret != "" { + if secret, derr := crypto.DecryptField(wh.Secret); derr == nil && secret != "" { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(payload) + req.Header.Set("X-FlowStack-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil))) + } + } + client := &http.Client{Timeout: 10 * time.Second} + resp, err := client.Do(req) + now := time.Now() + if err != nil { + database.DB.Model(&wh).Updates(map[string]any{"last_status": "failed", "last_triggered_at": &now}) + c.JSON(http.StatusOK, gin.H{"success": false, "message": "发送失败: " + err.Error()}) + return + } + defer resp.Body.Close() + status := "success" + if resp.StatusCode >= 300 { + status = "failed" + } + database.DB.Model(&wh).Updates(map[string]any{"last_status": status, "last_triggered_at": &now}) + c.JSON(http.StatusOK, gin.H{"success": resp.StatusCode < 300, "statusCode": resp.StatusCode, "message": fmt.Sprintf("响应 %d %s", resp.StatusCode, http.StatusText(resp.StatusCode))}) +} + +// NotifyWebhookEvent 流水线事件发生后通知订阅了该事件的 Webhook（异步调用） +func NotifyWebhookEvent(event string, data map[string]any) { + go func() { + var endpoints []model.WebhookEndpoint + database.DB.Where("active = ?", true).Find(&endpoints) + payload, _ := json.Marshal(map[string]any{"event": event, "timestamp": time.Now().Format(time.RFC3339), "data": data}) + client := &http.Client{Timeout: 10 * time.Second} + for _, wh := range endpoints { + if !strings.Contains(wh.Events, event) { + continue + } + req, err := http.NewRequest("POST", wh.URL, strings.NewReader(string(payload))) + if err != nil { + continue + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-FlowStack-Event", event) + if wh.Secret != "" { + if secret, derr := crypto.DecryptField(wh.Secret); derr == nil && secret != "" { + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(payload) + req.Header.Set("X-FlowStack-Signature", "sha256="+hex.EncodeToString(mac.Sum(nil))) + } + } + status := "success" + if resp, err := client.Do(req); err != nil || resp.StatusCode >= 300 { + status = "failed" + if resp != nil { + resp.Body.Close() + } + } + now := time.Now() + database.DB.Model(&wh).Updates(map[string]any{"last_status": status, "last_triggered_at": &now}) + } + }() +} + +// InboundPipelineHook POST /api/hooks/pipeline/:id?token= —— 入站触发（TriggerPanel 的 Webhook URL） +func InboundPipelineHook(hub *Hub) gin.HandlerFunc { + return func(c *gin.Context) { + var p model.Pipeline + if err := database.DB.First(&p, c.Param("id")).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"}) + return + } + token := c.Query("token") + if p.WebhookToken != "" && token != p.WebhookToken { + c.JSON(http.StatusUnauthorized, gin.H{"error": "token 校验失败"}) + return + } + if p.TriggerMode != "webhook" && p.TriggerMode != "git_push" { + c.JSON(http.StatusConflict, gin.H{"error": "流水线未启用 Webhook 触发"}) + return + } + go triggerPipelineRun(hub, p, "", "", "webhook", "webhook") + c.JSON(http.StatusOK, gin.H{"success": true, "message": "已触发流水线 " + p.Name}) + } +} diff --git a/backend/internal/handler/runs.go b/backend/internal/handler/runs.go index a105fbc..7091193 100644 --- a/backend/internal/handler/runs.go +++ b/backend/internal/handler/runs.go @@ -13,7 +13,7 @@ import ( "flowstack/internal/model" ) -// CreateRun 创建运行并启动模拟执行引擎 +// CreateRun POST /api/pipelines/:id/runs —— 创建运行并启动执行引擎 func CreateRun(hub *Hub) gin.HandlerFunc { return func(c *gin.Context) { var p model.Pipeline @@ -21,61 +21,82 @@ func CreateRun(hub *Hub) gin.HandlerFunc { c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"}) return } - var body struct { - Branch string `json:"branch"` - Env string `json:"env"` - TriggerBy string `json:"triggerBy"` + Branch string `json:"branch"` + Env string `json:"env"` + TriggerBy string `json:"triggerBy"` + TriggerType string `json:"triggerType"` } _ = c.ShouldBindJSON(&body) - if body.Branch == "" { - body.Branch = p.SourceBranch - } - if body.Env == "" { - body.Env = p.Env - } - if body.TriggerBy == "" { - body.TriggerBy = "admin" - } - var nodes []model.PipelineNode - database.DB.Where("pipeline_id = ?", p.ID).Find(&nodes) - if len(nodes) == 0 { - c.JSON(http.StatusBadRequest, gin.H{"error": "流水线没有节点，请先在画布中编排并保存"}) + run, err := triggerPipelineRun(hub, p, body.Branch, body.Env, body.TriggerBy, body.TriggerType) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } - var edges []model.PipelineEdge - database.DB.Where("pipeline_id = ?", p.ID).Find(&edges) - nodesJSON, _ := json.Marshal(nodes) - edgesJSON, _ := json.Marshal(edges) - - run := model.PipelineRun{ - RunID: fmt.Sprintf("pr-%d", time.Now().UnixMilli()), - PipelineID: p.ID, PipelineName: p.Name, - Branch: body.Branch, Env: body.Env, DeployType: p.DeployType, Type: p.Type, - TriggerBy: body.TriggerBy, TriggerType: "manual", - Status: "running", StartedAt: time.Now(), - NodesSnapshot: string(nodesJSON), EdgesSnapshot: string(edgesJSON), - } - if err := database.DB.Create(&run).Error; err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) - return - } - - // 按画布 X 坐标从左到右生成执行阶段 - sort.Slice(nodes, func(i, j int) bool { return nodes[i].PositionX < nodes[j].PositionX }) - stages := make([]model.RunStage, len(nodes)) - for i, n := range nodes { - stages[i] = model.RunStage{RunID: run.ID, NodeID: n.NodeID, Name: n.Name, Status: "pending"} - database.DB.Create(&stages[i]) - } - - go simulateRun(hub, run, stages) - c.JSON(http.StatusOK, run) } } +// triggerPipelineRun 统一的运行创建入口：校验节点 → 建 Run/Stages → 启动执行引擎 +func triggerPipelineRun(hub *Hub, p model.Pipeline, branch, env, triggerBy, triggerType string) (*model.PipelineRun, error) { + if branch == "" { + branch = p.SourceBranch + } + if env == "" { + env = p.Env + } + if triggerBy == "" { + triggerBy = "admin" + } + if triggerType == "" { + triggerType = "manual" + } + + var nodes []model.PipelineNode + database.DB.Where("pipeline_id = ?", p.ID).Find(&nodes) + if len(nodes) == 0 { + return nil, fmt.Errorf("流水线没有节点，请先在画布中编排并保存") + } + var edges []model.PipelineEdge + database.DB.Where("pipeline_id = ?", p.ID).Find(&edges) + nodesJSON, _ := json.Marshal(nodes) + edgesJSON, _ := json.Marshal(edges) + + run := model.PipelineRun{ + RunID: fmt.Sprintf("pr-%d", time.Now().UnixMilli()), + PipelineID: p.ID, PipelineName: p.Name, + Branch: branch, Env: env, DeployType: p.DeployType, Type: p.Type, + TriggerBy: triggerBy, TriggerType: triggerType, + Status: "running", StartedAt: time.Now(), + NodesSnapshot: string(nodesJSON), EdgesSnapshot: string(edgesJSON), + } + if err := database.DB.Create(&run).Error; err != nil { + return nil, err + } + + // jenkins 类型流水线走真实 Jenkins 执行引擎 + if p.Type == "jenkins" && p.JenkinsJobName != "" { + var server model.JenkinsServer + if err := database.DB.First(&server, p.JenkinsServerID).Error; err == nil && server.Enabled { + client := clientFromServer(server) + go jenkinsRunEngine(hub, run, client, p.JenkinsJobName) + return &run, nil + } + } + + // 按画布 X 坐标从左到右生成执行阶段 + sort.Slice(nodes, func(i, j int) bool { return nodes[i].PositionX < nodes[j].PositionX }) + stages := make([]model.RunStage, len(nodes)) + for i, n := range nodes { + stages[i] = model.RunStage{RunID: run.ID, NodeID: n.NodeID, Name: n.Name, Status: "pending"} + database.DB.Create(&stages[i]) + } + + go simulateRun(hub, run, stages) + return &run, nil +} + // simulateRun 模拟执行：逐阶段 pending→running→success，逐条写日志并 SSE 推送 func simulateRun(hub *Hub, run model.PipelineRun, stages []model.RunStage) { start := time.Now() @@ -106,6 +127,13 @@ func simulateRun(hub *Hub, run model.PipelineRun, stages []model.RunStage) { updates := map[string]any{"status": status, "duration_ms": time.Since(start).Milliseconds(), "completed_at": now} database.DB.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Updates(updates) hub.Publish(run.RunID, "run", gin.H{"runId": run.RunID, "status": status, "durationMs": time.Since(start).Milliseconds()}) + // 通知订阅了 pipeline.success / pipeline.failed 的出站 Webhook + if status == "success" || status == "failed" { + NotifyWebhookEvent("pipeline."+status, map[string]any{ + "runId": run.RunID, "pipeline": run.PipelineName, + "branch": run.Branch, "durationMs": time.Since(start).Milliseconds(), + }) + } } emit("info", "Pipeline 触发成功，开始调度执行引擎…", 0) @@ -270,3 +298,25 @@ func StreamLogs(hub *Hub) gin.HandlerFunc { } } } + +// RetryRun POST /api/runs/:runId/retry —— 以同一流水线与分支重新触发 +func RetryRun(hub *Hub) gin.HandlerFunc { + return func(c *gin.Context) { + var prev model.PipelineRun + if err := database.DB.Where("run_id = ?", c.Param("runId")).First(&prev).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "run not found"}) + return + } + var p model.Pipeline + if err := database.DB.First(&p, prev.PipelineID).Error; err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"}) + return + } + run, err := triggerPipelineRun(hub, p, prev.Branch, prev.Env, "admin", "manual") + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, run) + } +} diff --git a/backend/internal/jenkins/client.go b/backend/internal/jenkins/client.go new file mode 100644 index 0000000..1d3176c --- /dev/null +++ b/backend/internal/jenkins/client.go @@ -0,0 +1,313 @@ +// Package jenkins Jenkins REST 客户端（移植自 opsx-jenkins 生产实现）： +// crumb → BasicAuth → buildWithParameters → Location 队列 URL → 队列/构建轮询 → progressiveText 增量控制台。 +package jenkins + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +type Client struct { + BaseURL string + Username string + Token string + HTTP *http.Client +} + +func NewClient(baseURL, username, token string) *Client { + return &Client{ + BaseURL: strings.TrimRight(baseURL, "/"), + Username: username, + Token: token, + HTTP: &http.Client{Timeout: 30 * time.Second}, + } +} + +// JobInfo Job 概要（folder 已压平为 path/name 形式） +type JobInfo struct { + Name string `json:"name"` // 压平路径 folder/job + URL string `json:"url"` + Type string `json:"type"` // folder | freestyle | pipeline | multibranch + Status string `json:"status"` // success | failed | running | unknown +} + +func (c *Client) do(method, path string, form url.Values) (*http.Response, error) { + var body io.Reader + if form != nil { + body = strings.NewReader(form.Encode()) + } + req, err := http.NewRequest(method, c.BaseURL+path, body) + if err != nil { + return nil, err + } + req.SetBasicAuth(c.Username, c.Token) + if form != nil { + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + } + // CSRF crumb（未启用时 /crumbIssuer 返回 404，跳过） + if crumb, err := c.crumb(); err == nil { + req.Header.Set(crumb[0], crumb[1]) + } + return c.HTTP.Do(req) +} + +func (c *Client) crumb() ([2]string, error) { + req, _ := http.NewRequest("GET", c.BaseURL+"/crumbIssuer/api/json", nil) + req.SetBasicAuth(c.Username, c.Token) + resp, err := c.HTTP.Do(req) + if err != nil { + return [2]string{}, err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return [2]string{}, errors.New("crumb not enabled") + } + var out struct { + CrumbRequestField string `json:"crumbRequestField"` + Crumb string `json:"crumb"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil || out.Crumb == "" { + return [2]string{}, errors.New("no crumb") + } + return [2]string{out.CrumbRequestField, out.Crumb}, nil +} + +// Test 连通测试：访问 /api/json +func (c *Client) Test() error { + resp, err := c.do("GET", "/api/json", nil) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusUnauthorized { + return errors.New("认证失败：用户名或 Token 错误") + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("HTTP %d", resp.StatusCode) + } + return nil +} + +type rawJob struct { + Name string `json:"name"` + URL string `json:"url"` + Color string `json:"color"` + Jobs []rawJob `json:"jobs"` +} + +// colorToStatus Jenkins ball color → 状态 +func colorToStatus(color string) string { + switch { + case color == "": + return "unknown" + case strings.HasSuffix(color, "anime"): + return "running" + case strings.HasPrefix(color, "blue"): + return "success" + case strings.HasPrefix(color, "red"): + return "failed" + case strings.HasPrefix(color, "notbuilt"), strings.HasPrefix(color, "disabled"), strings.HasPrefix(color, "aborted"): + return "unknown" + default: + return "unknown" + } +} + +// ListJobs 递归 3 层拉取 Job 树并压平 folder +func (c *Client) ListJobs() ([]JobInfo, error) { + resp, err := c.do("GET", "/api/json?tree=jobs[name,url,color,jobs[name,url,color,jobs[name,url,color,jobs[name,url,color]]]]", nil) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("HTTP %d", resp.StatusCode) + } + var root struct { + Jobs []rawJob `json:"jobs"` + } + if err := json.NewDecoder(resp.Body).Decode(&root); err != nil { + return nil, err + } + var flat []JobInfo + var walk func(jobs []rawJob, prefix string) + walk = func(jobs []rawJob, prefix string) { + for _, j := range jobs { + full := j.Name + if prefix != "" { + full = prefix + "/" + j.Name + } + if len(j.Jobs) > 0 { + flat = append(flat, JobInfo{Name: full, URL: j.URL, Type: "folder", Status: "unknown"}) + walk(j.Jobs, full) + } else { + flat = append(flat, JobInfo{Name: full, URL: j.URL, Type: "pipeline", Status: colorToStatus(j.Color)}) + } + } + } + walk(root.Jobs, "") + return flat, nil +} + +// jobPathToURL folder/job → /job/folder/job/job/name +func jobPathToURL(jobPath string) string { + segs := strings.Split(jobPath, "/") + var b strings.Builder + for _, s := range segs { + b.WriteString("/job/") + b.WriteString(url.PathEscape(s)) + } + return b.String() +} + +// Trigger 触发构建，返回队列 URL（Location 头；拒绝回退 lastBuild 防串号） +func (c *Client) Trigger(jobPath string, params map[string]string) (string, error) { + path := jobPathToURL(jobPath) + var form url.Values + if len(params) > 0 { + form = url.Values{} + for k, v := range params { + form.Set(k, v) + } + path += "/buildWithParameters" + } else { + path += "/build" + } + resp, err := c.do("POST", path, form) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return "", fmt.Errorf("触发失败 HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + queueURL := resp.Header.Get("Location") + if queueURL == "" { + return "", errors.New("未返回队列地址（Location），请检查 Jenkins 版本与权限") + } + return queueURL, nil +} + +// WaitQueue 等待队列分配为具体构建，返回构建号与构建 URL +func (c *Client) WaitQueue(queueURL string, timeout time.Duration, emit func(string)) (int, string, error) { + deadline := time.Now().Add(timeout) + pollClient := &http.Client{Timeout: 15 * time.Second} + for time.Now().Before(deadline) { + req, _ := http.NewRequest("GET", queueURL+"api/json", nil) + req.SetBasicAuth(c.Username, c.Token) + resp, err := pollClient.Do(req) + if err == nil { + var q struct { + Cancelled bool `json:"cancelled"` + Executable *struct { + Number int `json:"number"` + URL string `json:"url"` + } `json:"executable"` + Why string `json:"why"` + } + if json.NewDecoder(resp.Body).Decode(&q) == nil { + resp.Body.Close() + if q.Cancelled { + return 0, "", errors.New("构建已被 Jenkins 取消（排队被拒绝）") + } + if q.Executable != nil { + return q.Executable.Number, strings.TrimRight(q.Executable.URL, "/"), nil + } + if q.Why != "" { + emit("排队中: " + q.Why) + } + } else { + resp.Body.Close() + } + } + time.Sleep(2 * time.Second) + } + return 0, "", errors.New("等待队列超时") +} + +// StreamConsole 增量拉取控制台日志（X-Text-Size/X-More-Data 推进），直到构建结束 +func (c *Client) StreamConsole(buildURL string, emit func(string)) { + offset := 0 + pollClient := &http.Client{Timeout: 15 * time.Second} + for { + req, _ := http.NewRequest("GET", fmt.Sprintf("%s/logText/progressiveText?start=%d", buildURL, offset), nil) + req.SetBasicAuth(c.Username, c.Token) + resp, err := pollClient.Do(req) + if err != nil { + emit("日志拉取失败: " + err.Error()) + return + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if len(body) > 0 { + for _, line := range strings.Split(strings.TrimRight(string(body), "\n"), "\n") { + if line != "" { + emit(line) + } + } + } + if newSize := parseInt(resp.Header.Get("X-Text-Size")); newSize > offset { + offset = newSize + } + if resp.Header.Get("X-More-Data") != "true" { + return + } + time.Sleep(2 * time.Second) + } +} + +// WaitResult 轮询构建结果：SUCCESS/FAILURE/ABORTED/null(进行中) +func (c *Client) WaitResult(buildURL string, timeout time.Duration) (string, error) { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + req, _ := http.NewRequest("GET", buildURL+"/api/json", nil) + req.SetBasicAuth(c.Username, c.Token) + resp, err := c.HTTP.Do(req) + if err == nil { + var b struct { + Building bool `json:"building"` + Result *string `json:"result"` + } + if json.NewDecoder(resp.Body).Decode(&b) == nil { + resp.Body.Close() + if !b.Building && b.Result != nil { + return *b.Result, nil + } + } else { + resp.Body.Close() + } + } + time.Sleep(3 * time.Second) + } + return "", errors.New("等待构建结果超时") +} + +func parseInt(s string) int { + n := 0 + for _, ch := range s { + if ch < '0' || ch > '9' { + return n + } + n = n*10 + int(ch-'0') + } + return n +} + +// MapResult Jenkins 结果 → FlowStack 运行状态 +func MapResult(r string) string { + switch r { + case "SUCCESS": + return "success" + case "ABORTED": + return "cancelled" + default: + return "failed" + } +} diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index de529b1..c0bd662 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -15,18 +15,27 @@ type User struct { // Pipeline 流水线定义 type Pipeline struct { - ID uint `gorm:"primaryKey" json:"id"` - Name string `gorm:"index" json:"name"` - Description string `json:"description"` - Type string `gorm:"default:native" json:"type"` // native | jenkins | tekton - SourceRepo string `json:"sourceRepo"` - SourceBranch string `gorm:"default:main" json:"sourceBranch"` - GitopsRepo string `json:"gitopsRepo"` - GitopsPath string `json:"gitopsPath"` - DeployType string `gorm:"default:host" json:"deployType"` // host | k8s | gitops - Env string `gorm:"default:staging" json:"env"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"index" json:"name"` + Description string `json:"description"` + Type string `gorm:"default:native" json:"type"` // native | jenkins | tekton + SourceRepo string `json:"sourceRepo"` + SourceBranch string `gorm:"default:main" json:"sourceBranch"` + GitopsRepo string `json:"gitopsRepo"` + GitopsPath string `json:"gitopsPath"` + DeployType string `gorm:"default:host" json:"deployType"` // host | k8s | gitops + Env string `gorm:"default:staging" json:"env"` + TriggerMode string `gorm:"default:manual" json:"triggerMode"` // manual | git_push | cron | webhook + CronExpr string `json:"cronExpr"` + WebhookToken string `json:"webhookToken"` + WatchBranches string `json:"watchBranches"` + Variables string `gorm:"type:text" json:"variables"` // JSON [{key,value,remark}] + CachePaths string `gorm:"type:text" json:"cachePaths"` // JSON [{path,key}] + NotifyWebhook string `json:"notifyWebhookURL"` + JenkinsJobName string `json:"jenkinsJobName"` + JenkinsServerID uint `json:"jenkinsServerId"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } // PipelineNode 画布节点（JSON 字段名与前端 React Flow 对齐） @@ -67,8 +76,11 @@ type PipelineRun struct { Type string `gorm:"default:native" json:"type"` TriggerBy string `gorm:"default:system" json:"triggerBy"` TriggerType string `gorm:"default:manual" json:"triggerType"` // manual|webhook|cron|batch - Status string `gorm:"default:pending" json:"status"` // pending|running|success|failed|cancelled + Status string `gorm:"default:pending" json:"status"` // pending|queued|running|success|failed|cancelled DurationMs int64 `json:"durationMs"` + BuildNumber int `json:"buildNumber"` + ExternalRunID string `json:"externalRunId"` + ExternalURL string `json:"externalUrl"` NodesSnapshot string `json:"nodesSnapshot"` EdgesSnapshot string `json:"edgesSnapshot"` StartedAt time.Time `json:"startedAt"` @@ -103,19 +115,22 @@ type LogEntry struct { // Application GitOps 应用 type Application struct { - ID uint `gorm:"primaryKey" json:"id"` - Name string `gorm:"uniqueIndex" json:"name"` - Repo string `json:"repo"` - Branch string `gorm:"default:main" json:"branch"` - Path string `json:"path"` - Cluster string `gorm:"default:default" json:"cluster"` - Namespace string `gorm:"default:default" json:"namespace"` - ManifestType string `gorm:"default:plain" json:"manifestType"` // helm|kustomize|plain - SyncPolicy string `gorm:"default:manual" json:"syncPolicy"` // manual|automated - AutoPrune bool `json:"autoPrune"` - SelfHeal bool `json:"selfHeal"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"uniqueIndex" json:"name"` + Repo string `json:"repo"` + Branch string `gorm:"default:main" json:"branch"` + Path string `json:"path"` + Cluster string `gorm:"default:default" json:"cluster"` + Namespace string `gorm:"default:default" json:"namespace"` + ManifestType string `gorm:"default:plain" json:"manifestType"` // helm|kustomize|plain + SyncPolicy string `gorm:"default:manual" json:"syncPolicy"` // manual|automated + AutoPrune bool `json:"autoPrune"` + SelfHeal bool `json:"selfHeal"` + SyncStatus string `gorm:"default:Synced" json:"syncStatus"` // Synced|OutOfSync|Syncing + HealthStatus string `gorm:"default:Healthy" json:"healthStatus"` // Healthy|Progressing|Degraded + LastSyncAt *time.Time `json:"lastSyncAt"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` } // Plugin 插件 @@ -133,3 +148,125 @@ type Plugin struct { Config map[string]interface{} `gorm:"serializer:json" json:"config"` CreatedAt time.Time `json:"createdAt"` } + +// Host 主机资源 +type Host struct { + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"uniqueIndex" json:"name"` + IP string `json:"ip"` + SSHPort int `gorm:"default:22" json:"port"` + Username string `gorm:"default:root" json:"username"` + Tags string `gorm:"type:text" json:"tags"` // JSON ["production","web"] + Status string `gorm:"default:offline" json:"status"` // online|offline + OS string `json:"os"` + Remark string `json:"remark"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// K8sCluster 集群资源 +type K8sCluster struct { + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"uniqueIndex" json:"name"` + Context string `json:"context"` + APIServer string `json:"apiServer"` + Version string `json:"version"` + Kubeconfig string `gorm:"type:text" json:"-"` // enc:v1: 加密，不回显 + KubeconfigSet bool `gorm:"-" json:"kubeconfigSet"` + NodeCount int `json:"nodes"` + Namespaces string `gorm:"type:text" json:"namespaces"` // JSON [] + Status string `gorm:"default:disconnected" json:"status"` // connected|disconnected + LastHeartbeat *time.Time `json:"lastHeartbeat"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// Credential 凭证（内容 AES-GCM 加密，只写不读） +type Credential struct { + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"uniqueIndex" json:"name"` + Type string `gorm:"index" json:"type"` // ssh|password|docker|git|kubeconfig|aksk + Scope string `gorm:"default:全局" json:"scope"` + Content string `gorm:"type:text" json:"-"` // enc:v1: 加密 JSON + ContentSet bool `gorm:"-" json:"contentSet"` + Summary string `json:"summary"` // 脱敏摘要 + UsageCount int `gorm:"default:0" json:"usageCount"` + LastUsedAt *time.Time `json:"lastUsedAt"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// BuildNode 构建节点 +type BuildNode struct { + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"uniqueIndex" json:"name"` + IP string `json:"ip"` + Labels string `gorm:"type:text" json:"labels"` // JSON ["docker","go"] + MaxConcurrency int `gorm:"default:2" json:"maxConcurrency"` + WorkspacePath string `gorm:"default:/data/flowstack-workspace" json:"workspacePath"` + Enabled bool `gorm:"default:true" json:"enabled"` + Status string `gorm:"default:offline" json:"status"` // online|offline|busy + CurrentJobs int `gorm:"default:0" json:"currentJobs"` + LastHeartbeat *time.Time `json:"lastHeartbeat"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// DeployTool 构建工具（检测命令真实执行；安装为保守策略：仅检测不下载） +type DeployTool struct { + ID uint `gorm:"primaryKey" json:"id"` + ToolKey string `gorm:"uniqueIndex" json:"key"` + DisplayName string `json:"name"` + Category string `gorm:"default:build" json:"category"` // build|deploy + Status string `gorm:"default:not-installed" json:"status"` // not-installed|installed|installing|failed + Version string `json:"version"` // 默认/期望版本 + InstalledVersion string `json:"installedVersion"` // 真实探测到的版本 + InstallMode string `gorm:"default:lazy" json:"installMode"` // preinstalled|lazy|disabled + Path string `json:"path"` + CheckCommand string `gorm:"type:text" json:"checkCommand"` // 真实版本探测命令 + Enabled bool `gorm:"default:true" json:"enabled"` + LastCheck *time.Time `json:"lastCheck"` + ErrorMsg string `json:"errorMsg"` + SortOrder int `json:"sortOrder"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// WebhookEndpoint 出站 Webhook（HMAC-SHA256 签名） +type WebhookEndpoint struct { + ID uint `gorm:"primaryKey" json:"id"` + Name string `json:"name"` + URL string `json:"url"` + Secret string `gorm:"type:text" json:"-"` // enc:v1: 加密 + SecretSet bool `gorm:"-" json:"secretSet"` + Events string `gorm:"type:text" json:"events"` // JSON ["pipeline.success",...] + Active bool `gorm:"default:true" json:"active"` + LastTriggeredAt *time.Time `json:"lastTriggered"` + LastStatus string `gorm:"default:none" json:"lastStatus"` // success|failed|none + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// JenkinsServer Jenkins 实例 +type JenkinsServer struct { + ID uint `gorm:"primaryKey" json:"id"` + Name string `gorm:"uniqueIndex" json:"name"` + BaseURL string `json:"baseUrl"` + Username string `json:"username"` + Token string `gorm:"type:text" json:"-"` // enc:v1: 加密 + TokenSet bool `gorm:"-" json:"tokenSet"` + Enabled bool `gorm:"default:true" json:"enabled"` + Remark string `json:"remark"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +// SystemSetting 系统设置（单行，ID 恒为 1） +type SystemSetting struct { + ID uint `gorm:"primaryKey" json:"id"` + Theme string `gorm:"type:text" json:"theme"` // JSON {mode,primaryColor,...} + Notification string `gorm:"type:text" json:"notification"` // JSON {email,browser,...} + ToolMode string `gorm:"default:local" json:"toolMode"` // local|jenkins + Extra string `gorm:"type:text" json:"extra"` + UpdatedAt time.Time `json:"updatedAt"` +} diff --git a/backend/main.go b/backend/main.go index d44d608..0d788be 100644 --- a/backend/main.go +++ b/backend/main.go @@ -51,18 +51,24 @@ func main() { { api.POST("/auth/login", handler.Login) + api.GET("/summary", handler.GetSummary) + api.GET("/settings", handler.GetSettings) + api.PUT("/settings", handler.UpdateSettings) + api.GET("/pipelines", handler.ListPipelines) api.POST("/pipelines", handler.CreatePipeline) - api.GET("/pipelines/:id", handler.GetPipeline) + api.GET("/pipelines/:id", handler.GetPipelineDetail) api.PUT("/pipelines/:id", handler.UpdatePipeline) api.DELETE("/pipelines/:id", handler.DeletePipeline) api.PUT("/pipelines/:id/nodes", handler.SaveNodes) api.PUT("/pipelines/:id/edges", handler.SaveEdges) api.POST("/pipelines/:id/runs", handler.CreateRun(hub)) + api.POST("/hooks/pipeline/:id", handler.InboundPipelineHook(hub)) api.GET("/runs", handler.ListRuns) api.GET("/runs/:runId", handler.GetRun) api.GET("/runs/:runId/logs/stream", handler.StreamLogs(hub)) + api.POST("/runs/:runId/retry", handler.RetryRun(hub)) api.GET("/gitops/apps", handler.ListGitOpsApps) api.GET("/gitops/argo/apps", handler.ArgoList) @@ -75,6 +81,48 @@ func main() { api.POST("/plugins/install", handler.InstallPlugin) api.POST("/plugins/:id/toggle", handler.TogglePlugin) api.DELETE("/plugins/:id", handler.UninstallPlugin) + + api.GET("/hosts", handler.ListHosts) + api.POST("/hosts", handler.CreateHost) + api.PUT("/hosts/:id", handler.UpdateHost) + api.DELETE("/hosts/:id", handler.DeleteHost) + api.POST("/hosts/:id/test", handler.TestHost) + + api.GET("/clusters", handler.ListClusters) + api.POST("/clusters", handler.CreateCluster) + api.PUT("/clusters/:id", handler.UpdateCluster) + api.DELETE("/clusters/:id", handler.DeleteCluster) + api.POST("/clusters/:id/test", handler.TestCluster) + + api.GET("/credentials", handler.ListCredentials) + api.POST("/credentials", handler.CreateCredential) + api.PUT("/credentials/:id", handler.UpdateCredential) + api.DELETE("/credentials/:id", handler.DeleteCredential) + + api.GET("/build-nodes", handler.ListBuildNodes) + api.POST("/build-nodes", handler.CreateBuildNode) + api.PUT("/build-nodes/:id", handler.UpdateBuildNode) + api.DELETE("/build-nodes/:id", handler.DeleteBuildNode) + api.POST("/build-nodes/:id/test", handler.TestBuildNode) + + api.GET("/tools", handler.ListTools) + api.PUT("/tools/:id", handler.UpdateTool) + api.POST("/tools/:id/check", handler.CheckTool) + api.POST("/tools/:id/install", handler.InstallTool) + + api.GET("/webhooks", handler.ListWebhooks) + api.POST("/webhooks", handler.CreateWebhook) + api.PUT("/webhooks/:id", handler.UpdateWebhook) + api.DELETE("/webhooks/:id", handler.DeleteWebhook) + api.POST("/webhooks/:id/test", handler.TestWebhook) + + api.GET("/jenkins-servers", handler.ListJenkinsServers) + api.POST("/jenkins-servers", handler.CreateJenkinsServer) + api.PUT("/jenkins-servers/:id", handler.UpdateJenkinsServer) + api.DELETE("/jenkins-servers/:id", handler.DeleteJenkinsServer) + api.POST("/jenkins-servers/:id/test", handler.TestJenkinsServer) + api.GET("/jenkins-servers/:id/jobs", handler.ListJenkinsJobs) + api.POST("/jenkins-servers/:id/import", handler.ImportJenkinsJobs) } distFS, _ := fs.Sub(staticFS, "static") diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index ec40d96..c82a691 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,5 +1,6 @@ import { useState, useCallback } from 'react'; import { Toaster } from 'sonner'; +import { useQuery } from '@tanstack/react-query'; import Sidebar from '@/components/Sidebar'; import TopBar from '@/components/TopBar'; import PipelineFlowEditor from '@/components/flow/PipelineFlowEditor'; @@ -19,7 +20,7 @@ import WebhookConfig from '@/pages/WebhookConfig'; import GitOpsDeploy from '@/pages/GitOpsDeploy'; import PluginManager from '@/pages/PluginManager'; import SystemSettings from '@/pages/SystemSettings'; -import { pipelineScenarios } from '@/data/pipelineData'; +import { apiClient } from '@/lib/api'; export type ModuleType = 'pipelines' | 'gitops' | 'batch-release' | 'release-records' | 'environments' | 'credentials' | 'build-nodes' | 'tools' | 'jenkins-import' | 'webhooks' | 'plugins' | 'settings'; export type EditorTab = '基本信息' | '流程配置' | '触发设置' | '变量和缓存' | '发布'; @@ -31,19 +32,27 @@ export default function App() { // Pipeline list/editor view toggle const [pipelineView, setPipelineView] = useState<'list' | 'editor'>('list'); - // Current editing pipeline ID for flow editor const [currentPipelineId, setCurrentPipelineId] = useState(1); - // Pipeline editor state (only used when in pipeline editor mode) - const [activeScenario, setActiveScenario] = useState<'go-backend' | 'vue-frontend'>('go-backend'); const [activeTab, setActiveTab] = useState('流程配置'); - const [isSimulating, setIsSimulating] = useState(false); - const [executionStatus, setExecutionStatus] = useState<'idle' | 'running' | 'success' | 'failed'>('idle'); + const [saveTick, setSaveTick] = useState(0); + const [runTick, setRunTick] = useState(0); - const scenario = pipelineScenarios[activeScenario]; + const { data: pipelines = [] } = useQuery({ + queryKey: ['pipelines'], + queryFn: apiClient.listPipelines, + }); - const handleScenarioChange = (key: 'go-backend' | 'vue-frontend') => { - setActiveScenario(key); - }; + // 当前流水线最近一次运行状态（TopBar 徽章） + const { data: currentRuns = [] } = useQuery({ + queryKey: ['runs', { pipelineId: currentPipelineId }], + queryFn: () => apiClient.listRuns({ pipelineId: currentPipelineId }), + refetchInterval: (query) => + query.state.data?.[0]?.status === 'running' || query.state.data?.[0]?.status === 'pending' ? 2000 : false, + }); + + const currentPipeline = pipelines.find((p: any) => p.id === currentPipelineId); + const latestStatus = currentRuns[0]?.status as string | undefined; + const isPipelineRunning = latestStatus === 'running' || latestStatus === 'pending'; const handleSelectPipeline = useCallback((id: string) => { setCurrentPipelineId(Number(id) || 1); @@ -51,13 +60,19 @@ export default function App() { setActiveTab('流程配置'); }, []); - const handleSimulateRun = useCallback(() => { - setIsSimulating(true); - setExecutionStatus('running'); - setTimeout(() => { - setIsSimulating(false); - setExecutionStatus('success'); - }, 3000); + const handlePipelineSwitch = useCallback((id: number) => { + setCurrentPipelineId(id); + setActiveTab('流程配置'); + }, []); + + const handleOnlySave = useCallback(() => { + if (activeTab !== '流程配置') setActiveTab('流程配置'); + setSaveTick((t) => t + 1); + }, [activeTab]); + + const handleSaveAndRun = useCallback(() => { + setActiveTab('流程配置'); + setRunTick((t) => t + 1); }, []); // Render content based on active module @@ -71,40 +86,42 @@ export default function App() { return (
({ id: p.id, name: p.name }))} + currentPipelineId={currentPipelineId} + onPipelineChange={handlePipelineSwitch} activeTab={activeTab} onTabChange={setActiveTab} - onSimulateRun={handleSimulateRun} - isSimulating={isSimulating} - scenarioKey={activeScenario} - onScenarioChange={handleScenarioChange} - executionStatus={executionStatus} + onOnlySave={handleOnlySave} + onSaveAndRun={handleSaveAndRun} + isRunning={isPipelineRunning} + executionStatus={(latestStatus as 'running' | 'success' | 'failed' | 'cancelled') || 'idle'} onBack={() => setPipelineView('list')} />
{activeTab === '流程配置' && (
- +
)} {activeTab === '基本信息' && (
- +
)} {activeTab === '触发设置' && (
- +
)} {activeTab === '变量和缓存' && (
- +
)} {activeTab === '发布' && ( -
- +
+
)}
@@ -121,7 +138,7 @@ export default function App() { case 'jenkins-import': return ; case 'batch-release': - return ; + return setActiveModule('release-records')} />; case 'release-records': return ; case 'webhooks': diff --git a/frontend/src/components/BasicInfoPanel.tsx b/frontend/src/components/BasicInfoPanel.tsx index cc75b5a..678f16e 100644 --- a/frontend/src/components/BasicInfoPanel.tsx +++ b/frontend/src/components/BasicInfoPanel.tsx @@ -1,270 +1,115 @@ -import { useState } from 'react'; -import { Bell, Info, TestTube } from 'lucide-react'; -import { Button } from '@/components/ui/button'; -import { Input } from '@/components/ui/input'; -import { Label } from '@/components/ui/label'; -import { Switch } from '@/components/ui/switch'; -import { Separator } from '@/components/ui/separator'; -import { RadioGroup, RadioGroupItem } from '@/components/ui/radio-group'; -import { Tooltip, TooltipContent, TooltipTrigger } from '@/components/ui/tooltip'; +import { useEffect, useState } from 'react'; +import { Info, Save, Loader2 } from 'lucide-react'; import { toast } from 'sonner'; +import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; +import { apiClient } from '@/lib/api'; -export default function BasicInfoPanel() { - const [pipelineName, setPipelineName] = useState('gateway-pre-serverless'); - const [description, setDescription] = useState(''); +interface Props { + pipelineId: number; +} - // Feishu notification - const [feishuEnabled, setFeishuEnabled] = useState(false); - const [feishuWebhook, setFeishuWebhook] = useState(''); - const [feishuTiming, setFeishuTiming] = useState('all'); +export default function BasicInfoPanel({ pipelineId }: Props) { + const queryClient = useQueryClient(); + const { data: detail } = useQuery({ + queryKey: ['pipeline', pipelineId], + queryFn: () => apiClient.getPipeline(pipelineId), + }); + const p = detail?.pipeline; - // DingTalk notification - const [dingtalkEnabled, setDingtalkEnabled] = useState(false); - const [dingtalkWebhook, setDingtalkWebhook] = useState(''); - const [dingtalkTiming, setDingtalkTiming] = useState('all'); + const [form, setForm] = useState({ name: '', description: '', sourceRepo: '', sourceBranch: '', deployType: 'host', env: 'staging', notifyWebhookURL: '' }); - const handleTestConnection = (type: 'feishu' | 'dingtalk') => { - toast.success(`${type === 'feishu' ? '飞书' : '钉钉'} Webhook 连接测试成功`, { - description: '通知服务可正常访问', - }); - }; + useEffect(() => { + if (p) { + setForm({ + name: p.name || '', description: p.description || '', + sourceRepo: p.sourceRepo || '', sourceBranch: p.sourceBranch || 'main', + deployType: p.deployType || 'host', env: p.env || 'staging', + notifyWebhookURL: p.notifyWebhookURL || '', + }); + } + }, [p?.id, p?.updatedAt]); - const timingOptions = [ - { value: 'all', label: '全部事件' }, - { value: 'success', label: '仅成功' }, - { value: 'failure', label: '仅失败' }, - ]; + const saveMutation = useMutation({ + mutationFn: (body: any) => apiClient.updatePipeline(pipelineId, body), + onSuccess: () => { + toast.success('基本信息已保存'); + queryClient.invalidateQueries({ queryKey: ['pipeline', pipelineId] }); + queryClient.invalidateQueries({ queryKey: ['pipelines'] }); + }, + onError: (err: Error) => toast.error('保存失败', { description: err.message }), + }); + + const set = (k: string, v: string) => setForm((f) => ({ ...f, [k]: v })); + + const inputStyle = { background: 'var(--panel-bg-secondary)', borderColor: 'var(--border-color-light)', color: 'var(--text-primary)' }; return ( -
- {/* Basic Information */} -
-
- -

- 基本信息 -

+
+
+
+
+ +

基本信息

+ {p?.type === 'jenkins' && ( + Jenkins 导入 + )} +
+
+ {p && 更新于 {new Date(p.updatedAt).toLocaleString('zh-CN', { hour12: false })}} + +
-
-
- - setPipelineName(e.target.value)} - placeholder="输入流水线名称" - className="h-9" - /> +
+
+ + set('name', e.target.value)} className="w-full px-3 py-2 text-[13px] rounded-md border outline-none" style={inputStyle} />
- -
- -