feat: 去 mock 化 —— 全模块接入真实数据 (参考 jianmu/opsx-jenkins/aiops)
后端（35+ 新路由）： - AES-256-GCM 字段加密(enc:v1:)，密钥自动生成 data/encryption.key - 新模型：Host/K8sCluster/Credential/BuildNode/DeployTool/WebhookEndpoint/ JenkinsServer/SystemSetting；Pipeline 扩展触发器/变量/缓存列， PipelineRun 扩展 buildNumber/externalUrl - 资源 CRUD + 真实测试端点：主机/集群/构建节点 TCP 拨号、 工具真实执行版本检测(保守安装策略)、Webhook HMAC-SHA256 签名发送、 入站流水线触发 /api/hooks/pipeline/:id - Jenkins 完整集成（移植 opsx-jenkins 生产客户端）：crumb/BasicAuth/ buildWithParameters/Location 队列/progressiveText 增量控制台， 服务器 CRUD+测试、Job 树拉取(folder 压平)、导入为流水线、 jenkins 类型流水线真实执行回传日志 - 流水线列表附带 lastRun、运行重试、summary 统计、系统设置单行表 - 运行结束触发出站 Webhook 通知；GitOps 同步状态真实推进 前端（14 页全部去 mock）： - 流水线列表/发布记录/批量发布/环境/凭证/构建节点/工具/Webhook/ 系统设置/Jenkins 导入/GitOps(资源树+差异) 全部接入真实 API - 编辑器 4 Tab 持久化（基本信息/触发设置含真实 Webhook URL/变量缓存/发布历史） - TopBar 真实流水线切换 + 仅保存/保存并运行 - Sidebar 徽标真实统计；删除未使用组件与 mock 数据文件
This commit is contained in:
Logan.Li committed
1 parent
04d1342b27
commit
26341c3e8a
37 files changed
+5702
-4683
No files matched your search
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user