983 lines
30 KiB
Go
983 lines
30 KiB
Go
package controller
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"ezops/model"
|
||
"ezops/pkg/remoteexec"
|
||
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
// ==================== 流水线执行引擎 V2（Redis 实时日志推送）====================
|
||
|
||
// LogEntry 日志条目
|
||
type LogEntry struct {
|
||
Time string `json:"time"`
|
||
Level string `json:"level"`
|
||
Stage string `json:"stage"`
|
||
Message string `json:"message"`
|
||
}
|
||
|
||
// StageStep 单个阶段定义
|
||
type StageStep struct {
|
||
Name string `json:"name"`
|
||
Type string `json:"type"` // checkout/build/test/image/deploy/verify
|
||
Status string `json:"status"`
|
||
}
|
||
|
||
var logger *zap.Logger
|
||
|
||
func init() {
|
||
logger, _ = zap.NewProduction()
|
||
}
|
||
|
||
// executePipeline 后台执行流水线（goroutine）— V2 Redis 版，支持 Master/SSH 构建节点 // Logan.Li
|
||
func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.PipelineRun, nodeID uint) {
|
||
defer h.releaseBuildNode(nodeID)
|
||
runID := run.ID
|
||
ctx := context.Background()
|
||
logKey := fmt.Sprintf("pipeline:run:%d:logs", runID)
|
||
pubCh := fmt.Sprintf("pipeline:run:%d", runID)
|
||
|
||
// 日志函数：同时写 Redis buffer + 发布 Pub/Sub
|
||
log := func(level, stage, msg string) {
|
||
entry := LogEntry{Time: time.Now().Format("15:04:05"), Level: level, Stage: stage, Message: msg}
|
||
data, _ := json.Marshal(entry)
|
||
|
||
// 追加到 Redis 列表
|
||
if h.cache != nil {
|
||
existing, _ := h.cache.Get(ctx, logKey)
|
||
var entries []LogEntry
|
||
if existing != "" {
|
||
json.Unmarshal([]byte(existing), &entries)
|
||
}
|
||
entries = append(entries, entry)
|
||
buf, _ := json.Marshal(entries)
|
||
h.cache.Set(ctx, logKey, string(buf), 30*time.Minute)
|
||
// Pub/Sub 实时推送（redka/local 无 Pub/Sub，WebSocket 走轮询）// Logan.Li
|
||
if h.cache.Capabilities().PubSub {
|
||
h.cache.Publish(ctx, pubCh, string(data))
|
||
}
|
||
}
|
||
}
|
||
|
||
updateStatus := func(status string, duration int) {
|
||
now := time.Now()
|
||
updates := map[string]interface{}{"status": status}
|
||
if duration > 0 {
|
||
updates["duration"] = duration
|
||
}
|
||
if status == "success" || status == "failed" || status == "cancelled" {
|
||
updates["finished_at"] = &now
|
||
}
|
||
h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Updates(updates)
|
||
// 推送状态变更
|
||
if h.cache != nil && h.cache.Capabilities().PubSub {
|
||
statusData, _ := json.Marshal(map[string]interface{}{"type": "status", "status": status, "duration": duration})
|
||
h.cache.Publish(ctx, pubCh, string(statusData))
|
||
}
|
||
}
|
||
|
||
updatePipelineStats := func(status string) {
|
||
ups := map[string]interface{}{"last_status": status, "last_run_at": time.Now(), "run_count": pipeline.RunCount + 1}
|
||
if status == "success" {
|
||
ups["success_count"] = pipeline.SuccessCount + 1
|
||
}
|
||
if status == "failed" {
|
||
ups["fail_count"] = pipeline.FailCount + 1
|
||
}
|
||
h.db.Model(&model.Pipeline{}).Where("id = ?", pipeline.ID).Updates(ups)
|
||
}
|
||
|
||
isCancelled := func() bool {
|
||
var current model.PipelineRun
|
||
h.db.First(¤t, runID)
|
||
return current.Status == "cancelled"
|
||
}
|
||
|
||
// 解析阶段
|
||
var stages []StageStep
|
||
if pipeline.Stages != "" {
|
||
json.Unmarshal([]byte(pipeline.Stages), &stages)
|
||
}
|
||
if len(stages) == 0 {
|
||
log("warn", "init", "没有配置阶段，流水线结束")
|
||
log("success", "完成", "流水线执行成功 (0s)")
|
||
updateStatus("success", 0)
|
||
updatePipelineStats("success")
|
||
h.dequeueNext(pipeline.ID)
|
||
h.flushLogsToDB(ctx, runID, logKey)
|
||
return
|
||
}
|
||
|
||
// 工作目录 — 按仓库名+分支+时间戳命名，方便区分不同构建
|
||
branch := run.Branch
|
||
if branch == "" {
|
||
branch = pipeline.Branch
|
||
}
|
||
gitURL := pipeline.Source
|
||
// 从 Git URL 提取仓库名 (e.g. https://gitee.com/attacker/demo-jar.git → demo-jar)
|
||
repoName := extractRepoName(gitURL)
|
||
if repoName == "" {
|
||
repoName = pipeline.Code
|
||
}
|
||
ts := time.Now()
|
||
tsDir := fmt.Sprintf("%d%02d%02d%02d%02d", ts.Year(), ts.Month(), ts.Day(), ts.Hour(), ts.Minute())
|
||
localWorkspace := fmt.Sprintf("/app/workspace/%s-%s", repoName, pipeline.Code)
|
||
workspace := localWorkspace
|
||
|
||
execCtx, err := h.buildExecCtx(nodeID, workspace)
|
||
if err != nil {
|
||
log("error", "init", fmt.Sprintf("构建执行上下文失败: %v", err))
|
||
updateStatus("failed", 0)
|
||
updatePipelineStats("failed")
|
||
h.dequeueNext(pipeline.ID)
|
||
h.flushLogsToDB(ctx, runID, logKey)
|
||
return
|
||
}
|
||
if nodeID > 0 {
|
||
var bn model.BuildNode
|
||
if h.db.First(&bn, nodeID).Error == nil {
|
||
workspace = remoteWorkspacePath(bn, repoName, pipeline.Code)
|
||
}
|
||
}
|
||
execCtx.Workspace = workspace
|
||
|
||
if err := h.prepareWorkspace(execCtx, log); err != nil {
|
||
log("error", "init", fmt.Sprintf("初始化工作区失败: %v", err))
|
||
updateStatus("failed", 0)
|
||
updatePipelineStats("failed")
|
||
h.dequeueNext(pipeline.ID)
|
||
h.flushLogsToDB(ctx, runID, logKey)
|
||
return
|
||
}
|
||
|
||
runUpdates := map[string]interface{}{
|
||
"workspace_path": workspace,
|
||
"build_node_name": execCtx.NodeName,
|
||
}
|
||
if nodeID > 0 {
|
||
runUpdates["build_node_id"] = nodeID
|
||
}
|
||
h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Updates(runUpdates)
|
||
|
||
imageTag := fmt.Sprintf("%s-%s", repoName, tsDir)
|
||
|
||
log("info", "init", fmt.Sprintf("执行节点: %s", execCtx.NodeName))
|
||
log("info", "init", fmt.Sprintf("触发方式: 手动 (admin)"))
|
||
log("info", "init", fmt.Sprintf("分支: %s | 镜像标签: %s", branch, imageTag))
|
||
|
||
totalStart := time.Now()
|
||
|
||
for _, stage := range stages {
|
||
if isCancelled() {
|
||
log("warn", stage.Name, "流水线已被停止")
|
||
updateStatus("cancelled", int(time.Since(totalStart).Seconds()))
|
||
updatePipelineStats("cancelled")
|
||
h.dequeueNext(pipeline.ID)
|
||
h.flushLogsToDB(ctx, runID, logKey)
|
||
return
|
||
}
|
||
|
||
log("info", stage.Name, fmt.Sprintf("===== 阶段: %s =====", stage.Name))
|
||
stageStart := time.Now()
|
||
|
||
var err error
|
||
switch stage.Type {
|
||
case "checkout":
|
||
err = h.runCheckout(execCtx, gitURL, branch, log)
|
||
case "build":
|
||
err = h.runBuild(execCtx, pipeline.Dockerfile, log)
|
||
case "test":
|
||
err = h.runTest(execCtx, log)
|
||
case "image":
|
||
err = h.runDockerBuild(execCtx, pipeline, imageTag, log)
|
||
case "deploy":
|
||
deployWS := execCtx.Workspace
|
||
if execCtx.Remote {
|
||
if syncErr := h.syncRemoteWorkspaceToLocal(execCtx, localWorkspace, log); syncErr != nil {
|
||
err = syncErr
|
||
} else {
|
||
deployWS = localWorkspace
|
||
}
|
||
}
|
||
if err == nil {
|
||
err = h.runDeploy(deployWS, pipeline, imageTag, log)
|
||
}
|
||
case "verify":
|
||
verifyWS := execCtx.Workspace
|
||
if execCtx.Remote {
|
||
verifyWS = localWorkspace
|
||
}
|
||
err = h.runVerify(verifyWS, log)
|
||
default:
|
||
log("info", stage.Name, "跳过未知阶段类型: "+stage.Type)
|
||
}
|
||
|
||
stageDur := time.Since(stageStart)
|
||
if err != nil {
|
||
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
|
||
}
|
||
log("success", stage.Name, fmt.Sprintf("✓ %s 完成 (%v)", stage.Name, stageDur.Round(time.Second)))
|
||
}
|
||
|
||
totalDur := int(time.Since(totalStart).Seconds())
|
||
updateStatus("success", totalDur)
|
||
log("success", "完成", fmt.Sprintf("流水线执行成功 (%ds)", totalDur))
|
||
|
||
updatePipelineStats("success")
|
||
h.dequeueNext(pipeline.ID)
|
||
h.flushLogsToDB(ctx, runID, logKey)
|
||
}
|
||
|
||
// flushLogsToDB Redis 日志批量写入 MySQL
|
||
func (h *PipelineHandler) flushLogsToDB(ctx context.Context, runID uint, logKey string) {
|
||
if h.cache == nil {
|
||
return
|
||
}
|
||
raw, err := h.cache.Get(ctx, logKey)
|
||
if err != nil || raw == "" {
|
||
return
|
||
}
|
||
var entries []LogEntry
|
||
if err := json.Unmarshal([]byte(raw), &entries); err != nil {
|
||
return
|
||
}
|
||
logsJSON, _ := json.Marshal(entries)
|
||
h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Update("logs", string(logsJSON))
|
||
// 清理 Redis 缓存
|
||
h.cache.Del(ctx, logKey)
|
||
}
|
||
|
||
// runCheckout git clone（Master 或远程构建节点） // Logan.Li
|
||
func (h *PipelineHandler) runCheckout(execCtx *PipelineExecCtx, url, branch string, logFn func(string, string, string)) error {
|
||
if url == "" {
|
||
logFn("warn", "代码检出", "没有配置 Git URL，跳过代码检出")
|
||
return nil
|
||
}
|
||
logFn("info", "代码检出", "> Git 拉取 "+url+" (分支: "+branch+")")
|
||
|
||
if !execCtx.Remote && h.toolInstaller != nil {
|
||
if err := h.toolInstaller.EnsureAvailable("git"); err != nil {
|
||
return fmt.Errorf("git 不可用（安装失败）: %v", err)
|
||
}
|
||
}
|
||
|
||
cloneCmd := fmt.Sprintf(
|
||
"git clone --depth 1 -b %s %s %s",
|
||
shellQuotePath(branch), shellQuotePath(url), shellQuotePath(execCtx.Workspace),
|
||
)
|
||
output, err := h.execOnCtx(execCtx, "", cloneCmd, 15*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(output), "\n") {
|
||
if line != "" {
|
||
logFn("info", "代码检出", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("git clone 失败: %v", err)
|
||
}
|
||
hashOut, _ := h.execOnCtx(execCtx, execCtx.Workspace, "git rev-parse --short HEAD", 30*time.Second)
|
||
logFn("success", "代码检出", fmt.Sprintf("已拉取 commit: %s", strings.TrimSpace(hashOut)))
|
||
return nil
|
||
}
|
||
|
||
// runBuild 构建（自动检测 Maven / npm / Go / Makefile，支持远程节点） // Logan.Li
|
||
func (h *PipelineHandler) runBuild(execCtx *PipelineExecCtx, dockerfile string, logFn func(string, string, string)) error {
|
||
workspace := execCtx.Workspace
|
||
buildDir := workspace
|
||
if dockerfile != "" {
|
||
if dir := dotdot(dockerfile); dir != "." && dir != "" {
|
||
testDir := workspace + "/" + dir
|
||
if h.pathExists(execCtx, testDir) {
|
||
buildDir = testDir
|
||
logFn("info", "构建", ">>> 切换到构建目录: "+dir)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 如果根目录没找到构建文件，扫描一级子目录
|
||
if !h.hasBuildFileCtx(execCtx, buildDir) {
|
||
type subMatch struct {
|
||
dir string
|
||
kind string // "maven", "npm", "go", "make"
|
||
}
|
||
var matches []subMatch
|
||
if execCtx.Remote {
|
||
out, _ := h.execOnCtx(execCtx, buildDir, `for d in */; do
|
||
[ "$d" = ".git/" ] && continue
|
||
[ -f "${d}pom.xml" ] && echo "maven:$d"
|
||
[ -f "${d}package.json" ] && echo "npm:$d"
|
||
[ -f "${d}go.mod" ] && echo "go:$d"
|
||
[ -f "${d}Makefile" ] && echo "make:$d"
|
||
done`, 60*time.Second)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line == "" || !strings.Contains(line, ":") {
|
||
continue
|
||
}
|
||
parts := strings.SplitN(line, ":", 2)
|
||
sub := strings.TrimSuffix(buildDir+"/"+strings.TrimSuffix(parts[1], "/"), "/")
|
||
matches = append(matches, subMatch{sub, parts[0]})
|
||
}
|
||
} else {
|
||
entries, err := os.ReadDir(buildDir)
|
||
if err == nil {
|
||
for _, e := range entries {
|
||
if !e.IsDir() || e.Name() == ".git" {
|
||
continue
|
||
}
|
||
sub := buildDir + "/" + e.Name()
|
||
if _, err := os.Stat(sub + "/pom.xml"); err == nil {
|
||
matches = append(matches, subMatch{sub, "maven"})
|
||
}
|
||
if _, err := os.Stat(sub + "/package.json"); err == nil {
|
||
matches = append(matches, subMatch{sub, "npm"})
|
||
}
|
||
if _, err := os.Stat(sub + "/go.mod"); err == nil {
|
||
matches = append(matches, subMatch{sub, "go"})
|
||
}
|
||
if _, err := os.Stat(sub + "/Makefile"); err == nil {
|
||
matches = append(matches, subMatch{sub, "make"})
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if len(matches) > 0 {
|
||
// 有多个匹配时列出，优先选择第一个
|
||
if len(matches) > 1 {
|
||
names := make([]string, len(matches))
|
||
for i, m := range matches {
|
||
names[i] = fmt.Sprintf("%s(%s)", m.dir[len(buildDir)+1:], m.kind)
|
||
}
|
||
logFn("info", "构建", ">>> 发现多个构建目录: "+strings.Join(names, ", "))
|
||
}
|
||
buildDir = matches[0].dir
|
||
logFn("info", "构建", ">>> 自动检测到构建目录: "+buildDir[len(workspace)+1:])
|
||
} else {
|
||
logFn("info", "构建", "未检测到构建文件 (pom.xml/go.mod/package.json)，跳过构建")
|
||
return nil
|
||
}
|
||
}
|
||
|
||
if h.pathExists(execCtx, buildDir+"/pom.xml") {
|
||
if !execCtx.Remote && h.toolInstaller != nil {
|
||
if err := h.toolInstaller.EnsureAvailable("jdk"); err != nil {
|
||
return fmt.Errorf("JDK 不可用（安装失败）: %v", err)
|
||
}
|
||
if err := h.toolInstaller.EnsureAvailable("maven"); err != nil {
|
||
return fmt.Errorf("Maven 不可用（安装失败）: %v", err)
|
||
}
|
||
}
|
||
logFn("info", "构建", ">>> Maven 构建 (mvn clean package -DskipTests)")
|
||
logFn("info", "构建", "🔨 正在下载依赖并编译...")
|
||
out, err := h.execOnCtx(execCtx, buildDir, "mvn clean package -DskipTests", 30*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line != "" {
|
||
logFn("info", "构建", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("Maven 构建失败: %v", err)
|
||
}
|
||
logFn("success", "构建", "Maven 构建成功 ✅")
|
||
return nil
|
||
}
|
||
|
||
if h.pathExists(execCtx, buildDir+"/package.json") {
|
||
if !execCtx.Remote && h.toolInstaller != nil {
|
||
if err := h.toolInstaller.EnsureAvailable("nodejs"); err != nil {
|
||
return fmt.Errorf("Node.js 不可用（安装失败）: %v", err)
|
||
}
|
||
}
|
||
logFn("info", "构建", ">>> npm 构建 (npm install && npm run build)")
|
||
logFn("info", "构建", "📦 正在安装依赖...")
|
||
out, err := h.execOnCtx(execCtx, buildDir, "npm install --registry=https://registry.npmmirror.com --legacy-peer-deps 2>&1", 30*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line != "" {
|
||
logFn("info", "构建", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("npm install 失败: %v", err)
|
||
}
|
||
logFn("info", "构建", "🏗️ 正在打包构建...")
|
||
out, err = h.execOnCtx(execCtx, buildDir, "npm run build 2>&1", 30*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line != "" {
|
||
logFn("info", "构建", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("npm build 失败: %v", err)
|
||
}
|
||
logFn("success", "构建", "npm 构建成功 ✅")
|
||
return nil
|
||
}
|
||
|
||
if h.pathExists(execCtx, buildDir+"/go.mod") {
|
||
if !execCtx.Remote && h.toolInstaller != nil {
|
||
if err := h.toolInstaller.EnsureAvailable("golang"); err != nil {
|
||
return fmt.Errorf("Go 不可用（安装失败）: %v", err)
|
||
}
|
||
}
|
||
logFn("info", "构建", ">>> Go 构建 (go build ./...)")
|
||
goEnv := "export GOROOT=/usr/local/go PATH=/usr/local/go/bin:$PATH GOPROXY=https://goproxy.cn,direct GOTOOLCHAIN=local && "
|
||
out, err := h.execOnCtx(execCtx, buildDir, goEnv+"go build ./...", 30*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line != "" {
|
||
logFn("info", "构建", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("Go 构建失败: %v", err)
|
||
}
|
||
logFn("success", "构建", "Go 构建成功")
|
||
return nil
|
||
}
|
||
|
||
logFn("info", "构建", "未检测到构建文件 (pom.xml/go.mod/package.json)，跳过构建")
|
||
return nil
|
||
}
|
||
|
||
// hasBuildFile 检查目录是否包含可识别的构建文件
|
||
func hasBuildFile(dir string) bool {
|
||
for _, f := range []string{"pom.xml", "package.json", "go.mod", "Makefile"} {
|
||
if _, err := os.Stat(dir + "/" + f); err == nil {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// runTest 执行测试 // Logan.Li
|
||
func (h *PipelineHandler) runTest(execCtx *PipelineExecCtx, logFn func(string, string, string)) error {
|
||
if h.pathExists(execCtx, execCtx.Workspace+"/pom.xml") {
|
||
logFn("info", "测试", ">>> Maven 测试 (mvn test)")
|
||
out, _ := h.execOnCtx(execCtx, execCtx.Workspace, "mvn test", 30*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line != "" {
|
||
logFn("info", "测试", line)
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// runDockerBuild Docker 构建 & 推送（支持远程节点，需节点已安装 Docker） // Logan.Li
|
||
func (h *PipelineHandler) runDockerBuild(execCtx *PipelineExecCtx, pipeline model.Pipeline, tag string, logFn func(string, string, string)) error {
|
||
workspace := execCtx.Workspace
|
||
dockerfile := pipeline.Dockerfile
|
||
if dockerfile == "" {
|
||
dockerfile = "Dockerfile"
|
||
}
|
||
|
||
fullImage := h.resolvePipelineImage(pipeline, tag)
|
||
logFn("info", "镜像打包", fmt.Sprintf("镜像地址: %s", fullImage))
|
||
|
||
buildDir := workspace
|
||
if dir := dotdot(dockerfile); dir != "." && dir != "" {
|
||
buildDir = workspace + "/" + dir
|
||
}
|
||
dfPath := dockerfile
|
||
if strings.Contains(dockerfile, "/") {
|
||
dfPath = dockerfile[strings.LastIndex(dockerfile, "/")+1:]
|
||
}
|
||
|
||
logFn("info", "镜像打包", fmt.Sprintf(">>> Docker 构建 (dockerfile: %s)", dfPath))
|
||
|
||
if !execCtx.Remote && h.toolInstaller != nil {
|
||
if err := h.toolInstaller.EnsureAvailable("docker-cli"); err != nil {
|
||
return fmt.Errorf("docker 不可用（安装失败）: %v", err)
|
||
}
|
||
}
|
||
|
||
buildCmd := fmt.Sprintf("docker build -t %s -f %s .", shellQuotePath(fullImage), shellQuotePath(dfPath))
|
||
out, err := h.execOnCtx(execCtx, buildDir, buildCmd, 60*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(out), "\n") {
|
||
if line != "" {
|
||
logFn("info", "镜像打包", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("Docker 构建失败: %v", err)
|
||
}
|
||
logFn("success", "镜像打包", "Docker 构建成功")
|
||
|
||
logFn("info", "镜像打包", ">>> 推送镜像: "+fullImage)
|
||
pushOut, err := h.execOnCtx(execCtx, buildDir, "docker push "+shellQuotePath(fullImage), 60*time.Minute)
|
||
for _, line := range strings.Split(strings.TrimSpace(pushOut), "\n") {
|
||
if line != "" {
|
||
logFn("info", "镜像打包", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("Docker 推送失败: %v", err)
|
||
}
|
||
logFn("success", "镜像打包", "镜像推送完成")
|
||
return nil
|
||
}
|
||
|
||
// runDeploy 部署：K8s 走 update-image CI；SSH 走 SCP + 远程命令 // Logan.Li
|
||
func (h *PipelineHandler) runDeploy(workspace string, pipeline model.Pipeline, tag string, logFn func(string, string, string)) error {
|
||
switch pipeline.DeployTarget {
|
||
case "k8s":
|
||
return h.deployToK8s(pipeline, tag, logFn)
|
||
case "ssh":
|
||
return h.deployToSSH(workspace, pipeline, logFn)
|
||
default:
|
||
return h.deployViaYaml(workspace, logFn)
|
||
}
|
||
}
|
||
|
||
// resolvePipelineImage 解析流水线构建镜像全名（与镜像打包阶段一致） // Logan.Li
|
||
func (h *PipelineHandler) resolvePipelineImage(pipeline model.Pipeline, tag string) string {
|
||
var registryURL, imageName string
|
||
if pipeline.RegistryCredentialID != nil {
|
||
var credential model.Credential
|
||
if err := h.db.Take(&credential, *pipeline.RegistryCredentialID).Error; err == nil {
|
||
registryURL = credential.Summary
|
||
}
|
||
}
|
||
appName := pipeline.Code
|
||
if pipeline.JobName != "" {
|
||
appName = pipeline.JobName
|
||
}
|
||
if imageName == "" {
|
||
imageName = appName
|
||
} else {
|
||
imageName += appName
|
||
}
|
||
if registryURL == "" {
|
||
registryURL = "registry.example.com"
|
||
}
|
||
return fmt.Sprintf("%s/%s:%s", registryURL, imageName, tag)
|
||
}
|
||
|
||
// deployToK8s 调用 K8s API 更新 Deployment 镜像（等同 curl CI update-image） // Logan.Li
|
||
func (h *PipelineHandler) deployToK8s(pipeline model.Pipeline, tag string, logFn func(string, string, string)) error {
|
||
if pipeline.K8sDeployment == "" || pipeline.K8sNamespace == "" {
|
||
return fmt.Errorf("未配置 K8s Deployment")
|
||
}
|
||
if pipeline.K8sClusterID == nil {
|
||
return fmt.Errorf("未配置 K8s 集群")
|
||
}
|
||
var cluster model.KubernetesCluster
|
||
if err := h.db.First(&cluster, *pipeline.K8sClusterID).Error; err != nil {
|
||
return fmt.Errorf("集群不存在: %v", err)
|
||
}
|
||
client, err := buildK8sClient(&cluster)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
fullImage := h.resolvePipelineImage(pipeline, tag)
|
||
logFn("info", "部署", fmt.Sprintf(">>> CI 更新镜像 %s/%s", pipeline.K8sNamespace, pipeline.K8sDeployment))
|
||
logFn("info", "部署", fmt.Sprintf("镜像: %s", fullImage))
|
||
logFn("info", "部署", fmt.Sprintf("API: POST /kubernetes/%s/namespace/%s/deployments/%s/update-image",
|
||
cluster.Code, pipeline.K8sNamespace, pipeline.K8sDeployment))
|
||
|
||
if err := client.UpdateDeploymentImage(pipeline.K8sNamespace, pipeline.K8sDeployment, pipeline.K8sContainerName, fullImage); err != nil {
|
||
return fmt.Errorf("更新 Deployment 镜像失败: %v", err)
|
||
}
|
||
logFn("success", "部署", "Deployment 镜像已更新，滚动发布已触发")
|
||
return nil
|
||
}
|
||
|
||
// deployToSSH 通过 SCP 上传构建产物并在关联服务器上执行部署命令 // Logan.Li
|
||
func (h *PipelineHandler) deployToSSH(workspace string, pipeline model.Pipeline, logFn func(string, string, string)) error {
|
||
if pipeline.ServerIDs == "" {
|
||
return fmt.Errorf("未关联部署主机")
|
||
}
|
||
var serverIDs []uint
|
||
if err := json.Unmarshal([]byte(pipeline.ServerIDs), &serverIDs); err != nil || len(serverIDs) == 0 {
|
||
return fmt.Errorf("server_ids 解析失败")
|
||
}
|
||
|
||
var servers []model.Server
|
||
if err := h.db.Where("id IN ?", serverIDs).Find(&servers).Error; err != nil {
|
||
return fmt.Errorf("加载服务器失败: %v", err)
|
||
}
|
||
if len(servers) == 0 {
|
||
return fmt.Errorf("关联的服务器不存在")
|
||
}
|
||
|
||
artifact := findDeployArtifact(workspace)
|
||
if artifact == "" {
|
||
return fmt.Errorf("工作区未找到可部署产物（jar/tar.gz/zip）")
|
||
}
|
||
|
||
remoteDir := strings.TrimSpace(pipeline.DeployRemotePath)
|
||
if remoteDir == "" {
|
||
remoteDir = fmt.Sprintf("/opt/apps/%s", pipeline.Code)
|
||
}
|
||
remotePath := remoteDir + "/" + filepath.Base(artifact)
|
||
|
||
deployScript := strings.TrimSpace(pipeline.DeployScript)
|
||
if deployScript == "" {
|
||
deployScript = fmt.Sprintf(
|
||
"mkdir -p %s && cd %s && (test -x ./restart.sh && ./restart.sh) || (systemctl restart %s 2>/dev/null) || echo '部署完成，请配置 deploy_script'",
|
||
remoteDir, remoteDir, pipeline.Code,
|
||
)
|
||
}
|
||
|
||
configs := make([]remoteexec.SSHConfig, 0, len(servers))
|
||
hostnames := make([]string, 0, len(servers))
|
||
for _, srv := range servers {
|
||
cfg, err := buildSSHConfig(h.db, srv)
|
||
if err != nil {
|
||
logFn("error", "部署", fmt.Sprintf("%s: %v", srv.Hostname, err))
|
||
continue
|
||
}
|
||
configs = append(configs, cfg)
|
||
hostnames = append(hostnames, srv.Hostname)
|
||
}
|
||
if len(configs) == 0 {
|
||
return fmt.Errorf("无可用 SSH 配置，请检查服务器凭据")
|
||
}
|
||
|
||
logFn("info", "部署", fmt.Sprintf(">>> SCP %s → %s（%s）", filepath.Base(artifact), remotePath, strings.Join(hostnames, ", ")))
|
||
uploadResults := remoteexec.BatchUpload(configs, artifact, remotePath, 5)
|
||
for _, r := range uploadResults {
|
||
if r.Success {
|
||
logFn("success", "部署", fmt.Sprintf("✓ %s 上传成功", r.Host))
|
||
} else {
|
||
logFn("error", "部署", fmt.Sprintf("✗ %s 上传失败: %s", r.Host, r.Error))
|
||
}
|
||
}
|
||
for _, r := range uploadResults {
|
||
if !r.Success {
|
||
return fmt.Errorf("部分主机 SCP 上传失败")
|
||
}
|
||
}
|
||
|
||
logFn("info", "部署", ">>> 远程执行部署命令")
|
||
execResults := remoteexec.BatchExec(configs, deployScript, 5*time.Minute, 5)
|
||
for _, r := range execResults {
|
||
for _, line := range strings.Split(r.Output, "\n") {
|
||
if strings.TrimSpace(line) != "" {
|
||
logFn("info", "部署", fmt.Sprintf("[%s] %s", r.Host, line))
|
||
}
|
||
}
|
||
if !r.Success {
|
||
errMsg := r.Error
|
||
if errMsg == "" {
|
||
errMsg = fmt.Sprintf("exit=%d", r.ExitCode)
|
||
}
|
||
logFn("error", "部署", fmt.Sprintf("✗ %s 部署失败: %s", r.Host, errMsg))
|
||
return fmt.Errorf("主机 %s 部署失败", r.Host)
|
||
}
|
||
logFn("success", "部署", fmt.Sprintf("✓ %s 部署完成", r.Host))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// deployViaYaml 兜底：仓库内 K8s YAML kubectl apply // Logan.Li
|
||
func (h *PipelineHandler) deployViaYaml(workspace string, logFn func(string, string, string)) error {
|
||
deployYamls := findFiles(workspace, ".yaml", ".yml")
|
||
var k8sFiles []string
|
||
for _, f := range deployYamls {
|
||
content, _ := os.ReadFile(f)
|
||
if strings.Contains(string(content), "kind:") &&
|
||
(strings.Contains(string(content), "Deployment") ||
|
||
strings.Contains(string(content), "Service") ||
|
||
strings.Contains(string(content), "StatefulSet")) {
|
||
k8sFiles = append(k8sFiles, f)
|
||
}
|
||
}
|
||
if len(k8sFiles) == 0 {
|
||
logFn("info", "部署", "未找到 K8s 部署文件，跳过部署")
|
||
return nil
|
||
}
|
||
for _, yamlFile := range k8sFiles {
|
||
logFn("info", "部署", ">>> kubectl apply -f "+yamlFile)
|
||
cmd := exec.Command("kubectl", "apply", "-f", yamlFile)
|
||
out, err := cmd.CombinedOutput()
|
||
for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
|
||
if line != "" {
|
||
logFn("info", "部署", line)
|
||
}
|
||
}
|
||
if err != nil {
|
||
return fmt.Errorf("kubectl apply 失败: %v", err)
|
||
}
|
||
}
|
||
logFn("success", "部署", "部署完成")
|
||
return nil
|
||
}
|
||
|
||
// findDeployArtifact 在工作区查找可 SCP 的构建产物 // Logan.Li
|
||
func findDeployArtifact(workspace string) string {
|
||
priorityDirs := []string{"target", "build/libs", "dist", "bin"}
|
||
priorityExts := []string{".jar", ".tar.gz", ".tgz", ".zip", ".war"}
|
||
|
||
var bestPath string
|
||
var bestTime time.Time
|
||
|
||
checkFile := func(path string) {
|
||
info, err := os.Stat(path)
|
||
if err != nil || info.IsDir() {
|
||
return
|
||
}
|
||
if info.ModTime().After(bestTime) {
|
||
bestTime = info.ModTime()
|
||
bestPath = path
|
||
}
|
||
}
|
||
|
||
for _, dir := range priorityDirs {
|
||
fullDir := filepath.Join(workspace, dir)
|
||
entries, err := os.ReadDir(fullDir)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
for _, e := range entries {
|
||
if e.IsDir() {
|
||
continue
|
||
}
|
||
name := e.Name()
|
||
for _, ext := range priorityExts {
|
||
if strings.HasSuffix(name, ext) {
|
||
checkFile(filepath.Join(fullDir, name))
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if bestPath != "" {
|
||
return bestPath
|
||
}
|
||
|
||
// 根目录兜底扫描
|
||
for _, ext := range priorityExts {
|
||
for _, f := range findFiles(workspace, ext) {
|
||
checkFile(f)
|
||
}
|
||
}
|
||
return bestPath
|
||
}
|
||
|
||
// runVerify 部署后验证
|
||
func (h *PipelineHandler) runVerify(workspace string, logFn func(string, string, string)) error {
|
||
logFn("info", "验证", ">>> 部署后健康检查")
|
||
cmd := exec.Command("kubectl", "get", "pods", "-o", "wide")
|
||
cmd.Dir = workspace
|
||
out, err := cmd.CombinedOutput()
|
||
if err != nil {
|
||
logFn("warn", "验证", fmt.Sprintf("kubectl get pods 失败: %v", err))
|
||
}
|
||
for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
|
||
if line != "" {
|
||
logFn("info", "验证", line)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ======== helpers ========
|
||
|
||
func dotdot(path string) string {
|
||
i := strings.LastIndex(path, "/")
|
||
if i < 0 {
|
||
return "."
|
||
}
|
||
return path[:i]
|
||
}
|
||
|
||
// extractRepoName 从 Git URL 提取仓库名
|
||
// e.g. https://gitee.com/attacker/demo-jar.git → demo-jar
|
||
func extractRepoName(url string) string {
|
||
// Remove trailing .git
|
||
clean := strings.TrimSuffix(url, ".git")
|
||
// Get last path segment
|
||
idx := strings.LastIndex(clean, "/")
|
||
if idx < 0 {
|
||
return ""
|
||
}
|
||
return clean[idx+1:]
|
||
}
|
||
|
||
// cleanOldWorkspaces 清理超过 maxAge 的 workspace 目录
|
||
func cleanOldWorkspaces(baseDir string, maxAge time.Duration) {
|
||
entries, err := os.ReadDir(baseDir)
|
||
if err != nil {
|
||
return
|
||
}
|
||
now := time.Now()
|
||
for _, e := range entries {
|
||
if !e.IsDir() {
|
||
continue
|
||
}
|
||
info, err := e.Info()
|
||
if err != nil {
|
||
continue
|
||
}
|
||
if now.Sub(info.ModTime()) > maxAge {
|
||
os.RemoveAll(baseDir + "/" + e.Name())
|
||
}
|
||
}
|
||
}
|
||
|
||
// ==================== 流水线队列引擎（Jenkins 风格并发控制）====================
|
||
|
||
var (
|
||
pipelineRunningCount = make(map[uint]int) // pipelineID → 当前运行数
|
||
pipelineQueues = make(map[uint][]uint) // pipelineID → 排队 runID 列表
|
||
queueMu sync.Mutex
|
||
)
|
||
|
||
// tryExecuteOrQueue 触发流水线时检查流水线并发 + 构建节点槽位 // Logan.Li
|
||
func (h *PipelineHandler) tryExecuteOrQueue(pipeline model.Pipeline, run model.PipelineRun) {
|
||
max := pipeline.MaxExecutors
|
||
if max <= 0 {
|
||
max = 2
|
||
}
|
||
|
||
queueMu.Lock()
|
||
running := pipelineRunningCount[pipeline.ID]
|
||
if running >= max {
|
||
pipelineQueues[pipeline.ID] = append(pipelineQueues[pipeline.ID], run.ID)
|
||
queueMu.Unlock()
|
||
h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued")
|
||
return
|
||
}
|
||
queueMu.Unlock()
|
||
|
||
nodeID, canNode := h.tryAcquireBuildNode(pipeline)
|
||
if !canNode {
|
||
queueMu.Lock()
|
||
pipelineQueues[pipeline.ID] = append(pipelineQueues[pipeline.ID], run.ID)
|
||
queueMu.Unlock()
|
||
h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued")
|
||
return
|
||
}
|
||
|
||
queueMu.Lock()
|
||
pipelineRunningCount[pipeline.ID] = running + 1
|
||
queueMu.Unlock()
|
||
|
||
h.markRunStarting(run, nodeID)
|
||
go h.executePipeline(pipeline, run, nodeID)
|
||
}
|
||
|
||
// markRunStarting 标记运行开始并记录执行节点 // Logan.Li
|
||
func (h *PipelineHandler) markRunStarting(run model.PipelineRun, nodeID uint) {
|
||
updates := map[string]interface{}{"status": "running", "build_node_name": "Master"}
|
||
if nodeID > 0 {
|
||
updates["build_node_id"] = nodeID
|
||
var node model.BuildNode
|
||
if h.db.First(&node, nodeID).Error == nil {
|
||
updates["build_node_name"] = node.Name
|
||
}
|
||
}
|
||
h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Updates(updates)
|
||
}
|
||
|
||
// 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 model.Pipeline
|
||
if err := h.db.First(&pipeline, pipelineID).Error; err != nil {
|
||
return
|
||
}
|
||
var run model.PipelineRun
|
||
if err := h.db.First(&run, nextRunID).Error; err != nil {
|
||
return
|
||
}
|
||
|
||
nodeID, canNode := h.tryAcquireBuildNode(pipeline)
|
||
if !canNode {
|
||
queueMu.Lock()
|
||
pipelineQueues[pipelineID] = append([]uint{nextRunID}, pipelineQueues[pipelineID]...)
|
||
if pipelineRunningCount[pipelineID] > 0 {
|
||
pipelineRunningCount[pipelineID]--
|
||
}
|
||
queueMu.Unlock()
|
||
return
|
||
}
|
||
|
||
h.markRunStarting(run, nodeID)
|
||
go h.executePipeline(pipeline, run, nodeID)
|
||
}
|
||
|
||
// 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(&model.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)
|
||
if err != nil {
|
||
return result
|
||
}
|
||
for _, e := range entries {
|
||
full := dir + "/" + e.Name()
|
||
if e.IsDir() {
|
||
if e.Name() != ".git" {
|
||
result = append(result, findFiles(full, exts...)...)
|
||
}
|
||
} else {
|
||
for _, ext := range exts {
|
||
if strings.HasSuffix(e.Name(), ext) {
|
||
result = append(result, full)
|
||
break
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return result
|
||
}
|