feat: FlowStack 发布系统首个可运行版本 (gin + React)
- Go gin 后端:流水线/节点/连线/运行/日志/GitOps/插件 REST API, SQLite(GORM) 自动建表 + seed,模拟执行引擎 + SSE 实时日志推送, 前端产物 go:embed 嵌入单二进制,一个端口同时服务 API 与页面 - React 19 前端:画布式流水线编辑器(React Flow + Tekton 任务库)、 GitOps 部署、批量发布、发布记录等 14 个模块页面, tRPC 全部替换为 REST(TanStack Query) - 首次启动生成随机密码管理员账号(日志 + data/initial-admin-password.txt) - run.sh 一键构建启动(默认端口 9527,支持 start/stop/restart/build/dev)
This commit is contained in:
commit
04d1342b27
115 files changed
+23268
No files matched your search
@@ -0,0 +1,36 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"flowstack/internal/crypto"
|
||||
"flowstack/internal/database"
|
||||
"flowstack/internal/model"
|
||||
)
|
||||
|
||||
// Login POST /api/auth/login {username, password}
|
||||
func Login(c *gin.Context) {
|
||||
var body struct {
|
||||
Username string `json:"username" binding:"required"`
|
||||
Password string `json:"password" binding:"required"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "用户名和密码不能为空"})
|
||||
return
|
||||
}
|
||||
|
||||
var user model.User
|
||||
if err := database.DB.Where("username = ?", body.Username).First(&user).Error; err != nil {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "用户名或密码错误"})
|
||||
return
|
||||
}
|
||||
if !crypto.VerifyPassword(body.Password, user.Password) {
|
||||
c.JSON(http.StatusUnauthorized, gin.H{"error": "用户名或密码错误"})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"success": true, "username": user.Username, "nickname": user.Nickname, "role": user.Role,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"flowstack/internal/database"
|
||||
"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
|
||||
database.DB.Order("updated_at DESC").Find(&apps)
|
||||
c.JSON(http.StatusOK, apps)
|
||||
}
|
||||
|
||||
// ArgoList GET /api/gitops/argo/apps
|
||||
func ArgoList(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, argoApps)
|
||||
}
|
||||
|
||||
// ArgoSync POST /api/gitops/argo/sync {appName}
|
||||
func ArgoSync(c *gin.Context) {
|
||||
var body struct {
|
||||
AppName string `json:"appName" binding:"required"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "appName required"})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"success": true, "appName": body.AppName, "message": "Sync triggered"})
|
||||
}
|
||||
|
||||
// ArgoResourceTree GET /api/gitops/argo/:appName/tree
|
||||
func ArgoResourceTree(c *gin.Context) {
|
||||
name := c.Param("appName")
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"nodes": []gin.H{
|
||||
{"uid": "1", "name": name + "-deployment", "kind": "Deployment", "namespace": "aiops", "status": "healthy", "resourceVersion": "v1"},
|
||||
{"uid": "2", "name": name + "-svc", "kind": "Service", "namespace": "aiops", "status": "healthy", "resourceVersion": "v1"},
|
||||
{"uid": "3", "name": name + "-hpa", "kind": "HorizontalPodAutoscaler", "namespace": "aiops", "status": "healthy", "resourceVersion": "v2"},
|
||||
},
|
||||
"edges": []gin.H{{"source": "1", "target": "2"}, {"source": "1", "target": "3"}},
|
||||
})
|
||||
}
|
||||
|
||||
// ArgoDiff GET /api/gitops/argo/:appName/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"},
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Hub 按 runId 分组的 SSE 订阅中心
|
||||
type Hub struct {
|
||||
mu sync.Mutex
|
||||
subs map[string]map[chan []byte]struct{}
|
||||
}
|
||||
|
||||
func NewHub() *Hub {
|
||||
return &Hub{subs: make(map[string]map[chan []byte]struct{})}
|
||||
}
|
||||
|
||||
func (h *Hub) Subscribe(runId string) chan []byte {
|
||||
ch := make(chan []byte, 64)
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.subs[runId] == nil {
|
||||
h.subs[runId] = make(map[chan []byte]struct{})
|
||||
}
|
||||
h.subs[runId][ch] = struct{}{}
|
||||
return ch
|
||||
}
|
||||
|
||||
func (h *Hub) Unsubscribe(runId string, ch chan []byte) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if set := h.subs[runId]; set != nil {
|
||||
delete(set, ch)
|
||||
if len(set) == 0 {
|
||||
delete(h.subs, runId)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Publish 向 runId 的所有订阅者推送一条 SSE 帧(event + data)
|
||||
func (h *Hub) Publish(runId, event string, payload any) {
|
||||
data, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
frame := []byte(fmt.Sprintf("event: %s\ndata: %s\n\n", event, data))
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for ch := range h.subs[runId] {
|
||||
select {
|
||||
case ch <- frame:
|
||||
default: // 订阅者消费过慢则丢弃,避免阻塞模拟引擎
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"flowstack/internal/database"
|
||||
"flowstack/internal/model"
|
||||
)
|
||||
|
||||
// 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)
|
||||
}
|
||||
|
||||
// GetPipeline GET /api/pipelines/:id —— 详情含画布节点与连线
|
||||
func GetPipeline(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
|
||||
}
|
||||
var nodes []model.PipelineNode
|
||||
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)
|
||||
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,
|
||||
})
|
||||
}
|
||||
|
||||
// CreatePipeline POST /api/pipelines
|
||||
func CreatePipeline(c *gin.Context) {
|
||||
var input struct {
|
||||
Name string `json:"name" binding:"required"`
|
||||
Description string `json:"description"`
|
||||
Type string `json:"type"`
|
||||
SourceRepo string `json:"sourceRepo"`
|
||||
SourceBranch string `json:"sourceBranch"`
|
||||
DeployType string `json:"deployType"`
|
||||
Env string `json:"env"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "name 不能为空"})
|
||||
return
|
||||
}
|
||||
setDefault(&input.Type, "native")
|
||||
setDefault(&input.SourceBranch, "main")
|
||||
setDefault(&input.DeployType, "host")
|
||||
setDefault(&input.Env, "staging")
|
||||
|
||||
p := model.Pipeline{
|
||||
Name: input.Name, Description: input.Description, Type: input.Type,
|
||||
SourceRepo: input.SourceRepo, SourceBranch: input.SourceBranch,
|
||||
DeployType: input.DeployType, Env: input.Env,
|
||||
}
|
||||
if err := database.DB.Create(&p).Error; err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, p)
|
||||
}
|
||||
|
||||
// UpdatePipeline PUT /api/pipelines/:id
|
||||
func UpdatePipeline(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
|
||||
}
|
||||
var input map[string]any
|
||||
if err := c.ShouldBindJSON(&input); err != nil {
|
||||
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}
|
||||
updates := map[string]any{}
|
||||
for k, v := range input {
|
||||
if allowed[k] {
|
||||
updates[k] = v
|
||||
}
|
||||
}
|
||||
if len(updates) > 0 {
|
||||
updates["updated_at"] = time.Now()
|
||||
database.DB.Model(&p).Updates(updates)
|
||||
}
|
||||
c.JSON(http.StatusOK, p)
|
||||
}
|
||||
|
||||
// DeletePipeline DELETE /api/pipelines/:id —— 级联删除节点与连线
|
||||
func DeletePipeline(c *gin.Context) {
|
||||
id := c.Param("id")
|
||||
database.DB.Where("pipeline_id = ?", id).Delete(&model.PipelineEdge{})
|
||||
database.DB.Where("pipeline_id = ?", id).Delete(&model.PipelineNode{})
|
||||
database.DB.Delete(&model.Pipeline{}, id)
|
||||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||||
}
|
||||
|
||||
// SaveNodes PUT /api/pipelines/:id/nodes —— 全量覆盖保存
|
||||
func SaveNodes(c *gin.Context) {
|
||||
pipelineID := c.Param("id")
|
||||
var p model.Pipeline
|
||||
if err := database.DB.First(&p, pipelineID).Error; err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"})
|
||||
return
|
||||
}
|
||||
var body struct {
|
||||
Nodes []model.PipelineNode `json:"nodes"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
database.DB.Where("pipeline_id = ?", p.ID).Delete(&model.PipelineNode{})
|
||||
for i := range body.Nodes {
|
||||
body.Nodes[i].ID = 0
|
||||
body.Nodes[i].PipelineID = p.ID
|
||||
database.DB.Create(&body.Nodes[i])
|
||||
}
|
||||
database.DB.Model(&p).Update("updated_at", time.Now())
|
||||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||||
}
|
||||
|
||||
// SaveEdges PUT /api/pipelines/:id/edges —— 全量覆盖保存
|
||||
func SaveEdges(c *gin.Context) {
|
||||
pipelineID := c.Param("id")
|
||||
var p model.Pipeline
|
||||
if err := database.DB.First(&p, pipelineID).Error; err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"})
|
||||
return
|
||||
}
|
||||
var body struct {
|
||||
Edges []model.PipelineEdge `json:"edges"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
database.DB.Where("pipeline_id = ?", p.ID).Delete(&model.PipelineEdge{})
|
||||
for i := range body.Edges {
|
||||
body.Edges[i].ID = 0
|
||||
body.Edges[i].PipelineID = p.ID
|
||||
database.DB.Create(&body.Edges[i])
|
||||
}
|
||||
database.DB.Model(&p).Update("updated_at", time.Now())
|
||||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||||
}
|
||||
|
||||
func setDefault(v *string, def string) {
|
||||
if *v == "" {
|
||||
*v = def
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"flowstack/internal/database"
|
||||
"flowstack/internal/model"
|
||||
)
|
||||
|
||||
// availablePlugins 可安装插件市场(演示数据)
|
||||
var availablePlugins = []gin.H{
|
||||
{"pluginId": "gitops-sync", "name": "GitOps Sync", "description": "Argo CD integration", "version": "1.3.0", "author": "OpsX Team", "category": "部署", "installs": 3420},
|
||||
{"pluginId": "tekton-tasks", "name": "Tekton Tasks", "description": "Cloud-native CI tasks", "version": "2.1.0", "author": "CloudNative", "category": "构建", "installs": 5180},
|
||||
{"pluginId": "dark-theme", "name": "Dark Theme", "description": "Dark UI theme", "version": "1.1.0", "author": "DesignLab", "category": "主题", "installs": 8920},
|
||||
{"pluginId": "sonarqube", "name": "SonarQube Scanner", "description": "Code quality scan", "version": "3.2.0", "author": "CodeQuality", "category": "扫描", "installs": 2150},
|
||||
{"pluginId": "vault-secrets", "name": "Vault Secrets", "description": "HashiCorp Vault integration", "version": "1.6.0", "author": "SecurityTeam", "category": "安全", "installs": 980},
|
||||
{"pluginId": "matrix-build", "name": "Matrix Build", "description": "Multi-dimension parallel build", "version": "2.0.0", "author": "BuildMaster", "category": "构建", "installs": 1670},
|
||||
{"pluginId": "slack-notify", "name": "Slack Notify", "description": "Slack notifications", "version": "1.3.0", "author": "NotifyHub", "category": "通知", "installs": 2890},
|
||||
}
|
||||
|
||||
// ListPlugins GET /api/plugins
|
||||
func ListPlugins(c *gin.Context) {
|
||||
var list []model.Plugin
|
||||
database.DB.Order("created_at DESC").Find(&list)
|
||||
c.JSON(http.StatusOK, list)
|
||||
}
|
||||
|
||||
// AvailablePlugins GET /api/plugins/available
|
||||
func AvailablePlugins(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, availablePlugins)
|
||||
}
|
||||
|
||||
// InstallPlugin POST /api/plugins/install
|
||||
func InstallPlugin(c *gin.Context) {
|
||||
var body struct {
|
||||
PluginID string `json:"pluginId" binding:"required"`
|
||||
Name string `json:"name" binding:"required"`
|
||||
Description string `json:"description"`
|
||||
Version string `json:"version" binding:"required"`
|
||||
Author string `json:"author"`
|
||||
Category string `json:"category"`
|
||||
EntryPath string `json:"entryPath"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if body.Category == "" {
|
||||
body.Category = "general"
|
||||
}
|
||||
p := model.Plugin{
|
||||
PluginID: body.PluginID, Name: body.Name, Description: body.Description,
|
||||
Version: body.Version, Author: body.Author, Category: body.Category, EntryPath: body.EntryPath,
|
||||
Installed: true, Enabled: true, Config: map[string]any{},
|
||||
}
|
||||
if err := database.DB.Create(&p).Error; err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, p)
|
||||
}
|
||||
|
||||
// TogglePlugin POST /api/plugins/:id/toggle {enabled}
|
||||
func TogglePlugin(c *gin.Context) {
|
||||
var p model.Plugin
|
||||
if err := database.DB.First(&p, c.Param("id")).Error; err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "plugin not found"})
|
||||
return
|
||||
}
|
||||
var body struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&body); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
database.DB.Model(&p).Update("enabled", body.Enabled)
|
||||
c.JSON(http.StatusOK, gin.H{"success": true, "enabled": body.Enabled})
|
||||
}
|
||||
|
||||
// UninstallPlugin DELETE /api/plugins/:id
|
||||
func UninstallPlugin(c *gin.Context) {
|
||||
database.DB.Delete(&model.Plugin{}, c.Param("id"))
|
||||
c.JSON(http.StatusOK, gin.H{"success": true})
|
||||
}
|
||||
@@ -0,0 +1,272 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"flowstack/internal/database"
|
||||
"flowstack/internal/model"
|
||||
)
|
||||
|
||||
// CreateRun 创建运行并启动模拟执行引擎
|
||||
func CreateRun(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
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Branch string `json:"branch"`
|
||||
Env string `json:"env"`
|
||||
TriggerBy string `json:"triggerBy"`
|
||||
}
|
||||
_ = 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": "流水线没有节点,请先在画布中编排并保存"})
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// simulateRun 模拟执行:逐阶段 pending→running→success,逐条写日志并 SSE 推送
|
||||
func simulateRun(hub *Hub, run model.PipelineRun, stages []model.RunStage) {
|
||||
start := time.Now()
|
||||
emit := func(level, msg string, stageID uint) {
|
||||
entry := model.LogEntry{RunID: run.ID, StageID: stageID, Level: level, Message: msg, Source: "runner", Timestamp: time.Now()}
|
||||
database.DB.Create(&entry)
|
||||
hub.Publish(run.RunID, "log", entry)
|
||||
}
|
||||
setStage := func(stage *model.RunStage, status string) {
|
||||
now := time.Now()
|
||||
updates := map[string]any{"status": status}
|
||||
if status == "running" {
|
||||
stage.StartedAt = &now
|
||||
updates["started_at"] = now
|
||||
} else {
|
||||
stage.CompletedAt = &now
|
||||
updates["completed_at"] = now
|
||||
exit := 0
|
||||
stage.ExitCode = &exit
|
||||
updates["exit_code"] = exit
|
||||
}
|
||||
database.DB.Model(&model.RunStage{}).Where("id = ?", stage.ID).Updates(updates)
|
||||
stage.Status = status
|
||||
hub.Publish(run.RunID, "stage", gin.H{"stageId": stage.ID, "nodeId": stage.NodeID, "name": stage.Name, "status": status})
|
||||
}
|
||||
setRun := func(status string) {
|
||||
now := time.Now()
|
||||
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()})
|
||||
}
|
||||
|
||||
emit("info", "Pipeline 触发成功,开始调度执行引擎…", 0)
|
||||
for i := range stages {
|
||||
st := &stages[i]
|
||||
setStage(st, "running")
|
||||
for _, line := range stageScript(st) {
|
||||
emit("info", line, st.ID)
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
setStage(st, "success")
|
||||
}
|
||||
setRun("success")
|
||||
}
|
||||
|
||||
// stageScript 按节点类别生成贴近真实构建过程的日志
|
||||
func stageScript(st *model.RunStage) []string {
|
||||
switch st.Name {
|
||||
case "git-clone":
|
||||
return []string{
|
||||
fmt.Sprintf("$ git clone --depth=50 --branch=master %s", "https://gitee.com/attacker/aiops.git"),
|
||||
"Cloning into 'source'...",
|
||||
"HEAD is now at 8f3c21a feat: release v2.4.0",
|
||||
}
|
||||
case "unit-test":
|
||||
return []string{
|
||||
"$ go test -v -race ./...",
|
||||
"ok \tflowstack/internal/handler\t2.318s",
|
||||
"ok \tflowstack/internal/model\t0.402s",
|
||||
"coverage: 87.4% of statements",
|
||||
}
|
||||
case "kaniko-build":
|
||||
return []string{
|
||||
"$ kaniko --dockerfile Dockerfile --destination registry.example.com/aiops/aiops-backend:" + time.Now().Format("200601021504"),
|
||||
"INFO[0002] Retrieving image manifest golang:1.24-alpine",
|
||||
"INFO[0018] Executing 12 build steps",
|
||||
"INFO[0041] Pushing image to registry.example.com",
|
||||
}
|
||||
case "trivy-scan":
|
||||
return []string{
|
||||
"$ trivy image --severity HIGH,CRITICAL registry.example.com/aiops/aiops-backend:latest",
|
||||
"alpine (apk): 0 HIGH, 0 CRITICAL",
|
||||
"Scan completed: no critical vulnerabilities found",
|
||||
}
|
||||
case "kubectl-deploy":
|
||||
return []string{
|
||||
"$ kubectl set image deployment/aiops-backend app=registry.example.com/aiops/aiops-backend:" + time.Now().Format("200601021504") + " -n aiops",
|
||||
"deployment.apps/aiops-backend image updated",
|
||||
"Waiting for rollout: 3/3 updated",
|
||||
"deployment \"aiops-backend\" successfully rolled out",
|
||||
}
|
||||
case "npm build":
|
||||
return []string{
|
||||
"$ npm ci --registry=https://registry.npmmirror.com",
|
||||
"$ npm run build",
|
||||
"vite v7.x building for production...",
|
||||
"✓ built in 18.4s",
|
||||
}
|
||||
case "npm audit":
|
||||
return []string{
|
||||
"$ npm audit --audit-level=high",
|
||||
"found 0 vulnerabilities",
|
||||
}
|
||||
case "主机部署":
|
||||
return []string{
|
||||
"$ tar zcf dist.tar.gz dist/ && scp dist.tar.gz production-group:/home/admin/app/",
|
||||
"$ rsync -avz --delete /tmp/dist/ /usr/share/nginx/html/",
|
||||
"$ nginx -t && nginx -s reload",
|
||||
"2/2 主机部署完成",
|
||||
}
|
||||
case "刷新 CDN":
|
||||
return []string{
|
||||
"$ curl -X POST https://cdn.aliyuncs.com/api/refresh",
|
||||
`{"urls":["https://aiops.example.com/*"]}`,
|
||||
"RefreshTaskId: 202610271200****",
|
||||
}
|
||||
default:
|
||||
return []string{
|
||||
fmt.Sprintf("[%s] 开始执行…", st.Name),
|
||||
fmt.Sprintf("[%s] 执行完成 (exit 0)", st.Name),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ListRuns GET /api/runs?pipelineId=&status=
|
||||
func ListRuns(c *gin.Context) {
|
||||
q := database.DB.Model(&model.PipelineRun{}).Order("started_at DESC").Limit(100)
|
||||
if v := c.Query("pipelineId"); v != "" {
|
||||
q = q.Where("pipeline_id = ?", v)
|
||||
}
|
||||
if v := c.Query("status"); v != "" && v != "all" {
|
||||
q = q.Where("status = ?", v)
|
||||
}
|
||||
var runs []model.PipelineRun
|
||||
q.Find(&runs)
|
||||
c.JSON(http.StatusOK, runs)
|
||||
}
|
||||
|
||||
// GetRun GET /api/runs/:runId
|
||||
func GetRun(c *gin.Context) {
|
||||
var run model.PipelineRun
|
||||
if err := database.DB.Where("run_id = ?", c.Param("runId")).First(&run).Error; err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "run not found"})
|
||||
return
|
||||
}
|
||||
var stages []model.RunStage
|
||||
database.DB.Where("run_id = ?", run.ID).Order("id").Find(&stages)
|
||||
var logs []model.LogEntry
|
||||
database.DB.Where("run_id = ?", run.ID).Order("timestamp").Find(&logs)
|
||||
c.JSON(http.StatusOK, gin.H{"run": run, "stages": stages, "logs": logs})
|
||||
}
|
||||
|
||||
// StreamLogs GET /api/runs/:runId/logs/stream —— SSE:先回放历史日志再实时推送
|
||||
func StreamLogs(hub *Hub) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
runId := c.Param("runId")
|
||||
var run model.PipelineRun
|
||||
if err := database.DB.Where("run_id = ?", runId).First(&run).Error; err != nil {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "run not found"})
|
||||
return
|
||||
}
|
||||
|
||||
c.Header("Content-Type", "text/event-stream")
|
||||
c.Header("Cache-Control", "no-cache")
|
||||
c.Header("Connection", "keep-alive")
|
||||
c.Header("X-Accel-Buffering", "no")
|
||||
|
||||
flush := func() { if f, ok := c.Writer.(http.Flusher); ok { f.Flush() } }
|
||||
|
||||
fmt.Fprintf(c.Writer, "event: connected\ndata: {\"runId\":\"%s\"}\n\n", runId)
|
||||
flush()
|
||||
|
||||
// 回放历史日志
|
||||
var history []model.LogEntry
|
||||
database.DB.Where("run_id = ?", run.ID).Order("timestamp").Find(&history)
|
||||
for _, entry := range history {
|
||||
data, _ := json.Marshal(entry)
|
||||
fmt.Fprintf(c.Writer, "event: log\ndata: %s\n\n", data)
|
||||
}
|
||||
if len(history) > 0 {
|
||||
flush()
|
||||
}
|
||||
|
||||
ch := hub.Subscribe(runId)
|
||||
defer hub.Unsubscribe(runId, ch)
|
||||
|
||||
heartbeat := time.NewTicker(15 * time.Second)
|
||||
defer heartbeat.Stop()
|
||||
|
||||
ctx := c.Request.Context()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-heartbeat.C:
|
||||
fmt.Fprint(c.Writer, ": ping\n\n")
|
||||
flush()
|
||||
case frame := <-ch:
|
||||
c.Writer.Write(frame)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user