From 7bc5169e8f4c412eaa59e12bd392dcc789ddba6f Mon Sep 17 00:00:00 2001 From: attacker Date: Wed, 24 Jun 2026 05:02:43 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B5=81=E6=B0=B4=E7=BA=BF=E9=98=9F?= =?UTF-8?q?=E5=88=97=E5=BC=95=E6=93=8E=20=E2=80=94=20max=5Fexecutors?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=E6=8E=A7=E5=88=B6=20+=20Jenkins=E9=A3=8E?= =?UTF-8?q?=E6=A0=BC=E6=9E=84=E5=BB=BA=E9=98=9F=E5=88=97+=E5=8F=96?= =?UTF-8?q?=E6=B6=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/api/handlers/pipeline.go | 34 +++++- .../api/handlers/pipeline_executor.go | 95 ++++++++++++++- backend/internal/api/routes/routes.go | 3 + backend/internal/models/cicd.go | 9 ++ .../src/components/deploy/PipelineManager.tsx | 112 ++++++++++++++++++ 5 files changed, 249 insertions(+), 4 deletions(-) diff --git a/backend/internal/api/handlers/pipeline.go b/backend/internal/api/handlers/pipeline.go index 42fb327..ead13c4 100644 --- a/backend/internal/api/handlers/pipeline.go +++ b/backend/internal/api/handlers/pipeline.go @@ -3,6 +3,7 @@ package handlers import ( "net/http" "regexp" + "strconv" "time" "ezops/internal/models" @@ -234,6 +235,8 @@ func (h *PipelineHandler) CreatePipeline(c *gin.Context) { WebhookToken: req.WebhookToken, // 缓存 CachePaths: req.CachePaths, + // 并发控制 + MaxExecutors: req.MaxExecutors, Enabled: true, } @@ -315,6 +318,7 @@ func (h *PipelineHandler) UpdatePipeline(c *gin.Context) { if req.WebhookToken != "" { updates["webhook_token"] = req.WebhookToken } if req.CachePaths != "" { updates["cache_paths"] = req.CachePaths } if req.Enabled != nil { updates["enabled"] = req.Enabled } + if req.MaxExecutors > 0 { updates["max_executors"] = req.MaxExecutors } if err := h.db.Model(&pipeline).Updates(updates).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{ @@ -418,8 +422,8 @@ func (h *PipelineHandler) TriggerPipeline(c *gin.Context) { return } - // 后台异步执行流水线 - go h.executePipeline(pipeline, run) + // 通过队列引擎执行（尊重 max_executors 并发限制） + go h.tryExecuteOrQueue(pipeline, run) c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已触发", Data: run}) } @@ -548,3 +552,29 @@ func (h *PipelineHandler) WebhookTrigger(c *gin.Context) { }, }) } + +// GetPipelineQueue 查询流水线队列状态 +func (h *PipelineHandler) GetPipelineQueue(c *gin.Context) { + id := getPipelineID(c) + if id == 0 { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无效的流水线编码"}) + return + } + status := h.GetQueueStatus(uint(id)) + c.JSON(http.StatusOK, models.Response{Code: 200, Data: status}) +} + +// CancelQueueItem 取消排队中的构建 +func (h *PipelineHandler) CancelQueueItem(c *gin.Context) { + id := getPipelineID(c) + runID, _ := strconv.ParseUint(c.Param("runId"), 10, 64) + if id == 0 || runID == 0 { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误"}) + return + } + if err := h.CancelQueued(uint(id), uint(runID)); err != nil { + c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: err.Error()}) + return + } + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已取消"}) +} diff --git a/backend/internal/api/handlers/pipeline_executor.go b/backend/internal/api/handlers/pipeline_executor.go index 5341bf9..8512888 100644 --- a/backend/internal/api/handlers/pipeline_executor.go +++ b/backend/internal/api/handlers/pipeline_executor.go @@ -7,6 +7,7 @@ import ( "os" "os/exec" "strings" + "sync" "time" "ezops/internal/models" @@ -108,6 +109,7 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P log("success", "完成", "流水线执行成功 (0s)") updateStatus("success", 0) updatePipelineStats("success") + h.dequeueNext(pipeline.ID) h.flushLogsToDB(ctx, runID, logKey) return } @@ -144,6 +146,7 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P log("warn", stage.Name, "流水线已被停止") updateStatus("cancelled", int(time.Since(totalStart).Seconds())) updatePipelineStats("cancelled") + h.dequeueNext(pipeline.ID) h.flushLogsToDB(ctx, runID, logKey) return } @@ -174,6 +177,7 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P log("error", stage.Name, fmt.Sprintf("✗ %s 失败 (%v): %v", stage.Name, stageDur.Round(time.Second), err)) updateStatus("failed", int(time.Since(totalStart).Seconds())) updatePipelineStats("failed") + h.dequeueNext(pipeline.ID) h.flushLogsToDB(ctx, runID, logKey) return } @@ -182,10 +186,10 @@ func (h *PipelineHandler) executePipeline(pipeline models.Pipeline, run models.P totalDur := int(time.Since(totalStart).Seconds()) updateStatus("success", totalDur) - updatePipelineStats("success") log("success", "完成", fmt.Sprintf("流水线执行成功 (%ds)", totalDur)) - // 执行结束 → Redis 日志批量写入 MySQL + updatePipelineStats("success") + h.dequeueNext(pipeline.ID) h.flushLogsToDB(ctx, runID, logKey) } @@ -549,6 +553,93 @@ func cleanOldWorkspaces(baseDir string, maxAge time.Duration) { } } +// ==================== 流水线队列引擎（Jenkins 风格并发控制）==================== + +var ( + pipelineRunningCount = make(map[uint]int) // pipelineID → 当前运行数 + pipelineQueues = make(map[uint][]uint) // pipelineID → 排队 runID 列表 + queueMu sync.Mutex +) + +// tryExecuteOrQueue 触发流水线时检查并发限制 +// 如果当前运行数 < maxExecutors → 立即执行 +// 否则 → 加入队列 (status=queued) +func (h *PipelineHandler) tryExecuteOrQueue(pipeline models.Pipeline, run models.PipelineRun) { + max := pipeline.MaxExecutors + if max <= 0 { max = 1 } + + queueMu.Lock() + running := pipelineRunningCount[pipeline.ID] + if running < max { + pipelineRunningCount[pipeline.ID] = running + 1 + queueMu.Unlock() + // 立即执行 + h.db.Model(&models.PipelineRun{}).Where("id = ?", run.ID).Update("status", "running") + go h.executePipeline(pipeline, run) + return + } + // 加入队列 + pipelineQueues[pipeline.ID] = append(pipelineQueues[pipeline.ID], run.ID) + queueMu.Unlock() + h.db.Model(&models.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued") +} + +// dequeueNext 流水线运行结束后，从队列中取下一个执行 +func (h *PipelineHandler) dequeueNext(pipelineID uint) { + queueMu.Lock() + running := pipelineRunningCount[pipelineID] + if running > 0 { + pipelineRunningCount[pipelineID] = running - 1 + } + + queue := pipelineQueues[pipelineID] + if len(queue) == 0 { + queueMu.Unlock() + return + } + nextRunID := queue[0] + pipelineQueues[pipelineID] = queue[1:] + pipelineRunningCount[pipelineID] = pipelineRunningCount[pipelineID] + 1 + queueMu.Unlock() + + // 加载 pipeline + run + var pipeline models.Pipeline + if err := h.db.First(&pipeline, pipelineID).Error; err != nil { return } + var run models.PipelineRun + if err := h.db.First(&run, nextRunID).Error; err != nil { return } + + h.db.Model(&run).Update("status", "running") + go h.executePipeline(pipeline, run) +} + +// GetQueueStatus 查询流水线队列状态（供 API 使用） +func (h *PipelineHandler) GetQueueStatus(pipelineID uint) map[string]interface{} { + queueMu.Lock() + defer queueMu.Unlock() + running := pipelineRunningCount[pipelineID] + queued := pipelineQueues[pipelineID] + return map[string]interface{}{ + "running": running, + "queued": len(queued), + "queue_ids": queued, + } +} + +// CancelQueued 取消排队中的构建 +func (h *PipelineHandler) CancelQueued(pipelineID uint, runID uint) error { + queueMu.Lock() + defer queueMu.Unlock() + queue := pipelineQueues[pipelineID] + for i, id := range queue { + if id == runID { + pipelineQueues[pipelineID] = append(queue[:i], queue[i+1:]...) + h.db.Model(&models.PipelineRun{}).Where("id = ?", runID).Update("status", "cancelled") + return nil + } + } + return fmt.Errorf("run %d not in queue", runID) +} + func findFiles(dir string, exts ...string) []string { var result []string entries, err := os.ReadDir(dir) diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index 84d6785..220385e 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -474,6 +474,9 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c pipelineScoped.POST("/runs/:runId/stop", pipelineHandler.StopPipelineRun) pipelineScoped.POST("/runs/batch-delete", pipelineHandler.BatchDeleteRuns) pipelineScoped.GET("/runs/:runId/log", pipelineHandler.GetRunLog) + // 队列管理 + pipelineScoped.GET("/queue", pipelineHandler.GetPipelineQueue) + pipelineScoped.DELETE("/queue/:runId", pipelineHandler.CancelQueueItem) // 成员管理 pipelineScoped.GET("/members", pipelineHandler.ListPipelineMembers) pipelineScoped.POST("/members", pipelineHandler.AddPipelineMember) diff --git a/backend/internal/models/cicd.go b/backend/internal/models/cicd.go index 0e3a666..89e0f54 100644 --- a/backend/internal/models/cicd.go +++ b/backend/internal/models/cicd.go @@ -24,6 +24,9 @@ type Pipeline struct { DeployMode string `gorm:"size:16;default:branch;comment:部署模式 branch/tag" json:"deploy_mode"` DeployTarget string `gorm:"size:16;default:ssh;comment:部署目标 ssh/k8s" json:"deploy_target"` + // 并发控制 + MaxExecutors int `gorm:"default:1;comment:最大并发执行数(1=串行)" json:"max_executors"` + // SSH 部署目标服务器 (JSON: [1,2,3]) ServerIDs string `gorm:"type:text;comment:SSH服务器ID列表(JSON)" json:"server_ids,omitempty"` @@ -170,6 +173,9 @@ type CreatePipelineRequest struct { // 缓存路径 CachePaths string `json:"cache_paths"` + + // 并发控制 + MaxExecutors int `json:"max_executors"` } // UpdatePipelineRequest 更新流水线请求 @@ -203,6 +209,9 @@ type UpdatePipelineRequest struct { // 缓存 CachePaths string `json:"cache_paths"` Enabled *bool `json:"enabled"` + + // 并发控制 + MaxExecutors int `json:"max_executors"` } // TriggerPipelineRequest 触发流水线请求 diff --git a/frontend/src/components/deploy/PipelineManager.tsx b/frontend/src/components/deploy/PipelineManager.tsx index 7b9faf4..03b1810 100644 --- a/frontend/src/components/deploy/PipelineManager.tsx +++ b/frontend/src/components/deploy/PipelineManager.tsx @@ -84,6 +84,9 @@ export function PipelineManager() { // Cache paths const [newCachePaths, setNewCachePaths] = useState(''); + // Executor concurrency + const [newMaxExecutors, setNewMaxExecutors] = useState(1); + // Reference data const [repos, setRepos] = useState([]); const [registries, setRegistries] = useState([]); @@ -190,6 +193,7 @@ export function PipelineManager() { variables: newEnvVars.length > 0 ? JSON.stringify(newEnvVars) : undefined, // 缓存路径 cache_paths: newCachePaths || undefined, + max_executors: newMaxExecutors, }); toast.success('流水线已创建'); setNewOpen(false); resetForm(); @@ -461,6 +465,7 @@ export function PipelineManager() { {isExpanded && (
+ {isLoadingRuns ? (
) : pipelineRuns.length === 0 ? ( @@ -796,6 +801,26 @@ export function PipelineManager() { placeholder="node_modules, .m2, target" className="bg-[var(--dg-bg-secondary)] border-[var(--dg-border)] h-8 text-xs font-mono" />

多个路径用逗号分隔，相同的 Key 会命中已有缓存

+ + {/* ══════════ 并发执行器 ══════════ */} +
+ +
+ +
+

设为 1 则按顺序串行执行，超过 1 可并发构建

+
@@ -834,3 +859,90 @@ export function PipelineManager() {
); } + +// ══════════ 流水线队列面板 (Jenkins 风格) ══════════ +function PipelineQueuePanel({ pipeline, clusterId: _cid }: { pipeline: any; clusterId: string }) { + const [queue, setQueue] = useState(null); + const [loading, setLoading] = useState(false); + const code = pipeline.code || pipeline.id; + + const loadQueue = async () => { + if (!code) return; + setLoading(true); + try { + const res: any = await http.get(`/pipelines/${code}/queue`); + setQueue(res?.data || res); + } catch { setQueue(null); } + finally { setLoading(false); } + }; + + useEffect(() => { loadQueue(); const t = setInterval(loadQueue, 10000); return () => clearInterval(t); }, [code]); + + const handleCancel = async (runId: number) => { + try { + await http.delete(`/pipelines/${code}/queue/${runId}`); + toast.success('已取消'); + loadQueue(); + } catch { toast.error('取消失败'); } + }; + + const running = queue?.running || 0; + const queued = queue?.queued || 0; + const queueIds: number[] = queue?.queue_ids || []; + const max = pipeline.max_executors || 1; + const pendingRuns = queue?.pending_runs || []; + + if (loading && !queue) return null; + if (running === 0 && queued === 0) return null; + + return ( +
+ {/* 构建执行状态 */} +
+ 构建执行状态 + {running}/{max} + {queued > 0 && ( + + 构建队列 ({queued}) + + )} +
+ + {/* 排队中的构建 */} + {queued > 0 && pendingRuns.length > 0 && ( +
+ {pendingRuns.map((run: any) => ( +
+
+ + #{run.build_number || run.id} + {run.branch || '-'} + {run.triggered_by || '-'} +
+ +
+ ))} +
+ )} + + {/* 仅有队列无详情时用 ID 列表兜底 */} + {queued > 0 && pendingRuns.length === 0 && queueIds.length > 0 && ( +
+ {queueIds.map((id: number) => ( +
+
+ + Run #{id} +
+ +
+ ))} +
+ )} +
+ ); +}