package controller import ( "fmt" "os" "os/exec" "path/filepath" "strings" "sync" "time" "ezops/model" "ezops/pkg/remoteexec" ) // Master 内置节点 ID(0 表示在本机执行) // Logan.Li const masterNodeID uint = 0 var ( nodeRunningCount = make(map[uint]int) masterMaxExecutors = 2 nodeMu sync.Mutex ) // ensureBuildMasterConfig 确保 Master 配置行存在并加载到内存 // Logan.Li func (h *PipelineHandler) ensureBuildMasterConfig() model.BuildMasterConfig { var cfg model.BuildMasterConfig if err := h.db.First(&cfg).Error; err != nil { cfg = model.BuildMasterConfig{MaxExecutors: 2} h.db.Create(&cfg) } if cfg.MaxExecutors <= 0 { cfg.MaxExecutors = 2 } nodeMu.Lock() masterMaxExecutors = cfg.MaxExecutors nodeMu.Unlock() return cfg } // setMasterMaxExecutors 更新 Master 并发上限(内存 + DB) // Logan.Li func (h *PipelineHandler) setMasterMaxExecutors(n int) model.BuildMasterConfig { if n <= 0 { n = 1 } cfg := h.ensureBuildMasterConfig() cfg.MaxExecutors = n h.db.Model(&cfg).Update("max_executors", n) nodeMu.Lock() masterMaxExecutors = n nodeMu.Unlock() return cfg } // PipelineExecCtx 流水线执行上下文(Master 或远程 SSH 节点) // Logan.Li type PipelineExecCtx struct { Remote bool SSHCfg *remoteexec.SSHConfig Workspace string NodeID uint NodeName string } // tryAcquireBuildNode 尝试占用构建节点执行槽位 // Logan.Li func (h *PipelineHandler) tryAcquireBuildNode(pipeline model.Pipeline) (uint, bool) { nodeMu.Lock() defer nodeMu.Unlock() // 指定节点 if pipeline.BuildNodeID != nil && *pipeline.BuildNodeID > 0 { id := *pipeline.BuildNodeID var node model.BuildNode if h.db.First(&node, id).Error != nil || !node.Enabled { return h.acquireMasterLocked() } max := node.MaxExecutors if max <= 0 { max = 1 } if nodeRunningCount[id] < max { nodeRunningCount[id]++ return id, true } return id, false } // 自动调度:Master 优先,其次负载最低的远程节点 if id, ok := h.acquireMasterLocked(); ok { return id, true } var nodes []model.BuildNode h.db.Where("enabled = ?", true).Order("id ASC").Find(&nodes) bestID := uint(0) bestScore := 2.0 for _, n := range nodes { max := n.MaxExecutors if max <= 0 { max = 1 } score := float64(nodeRunningCount[n.ID]) / float64(max) if score < 1.0 && score <= bestScore { bestScore = score bestID = n.ID } } if bestID > 0 { nodeRunningCount[bestID]++ return bestID, true } return 0, false } func (h *PipelineHandler) acquireMasterLocked() (uint, bool) { if nodeRunningCount[masterNodeID] < masterMaxExecutors { nodeRunningCount[masterNodeID]++ return masterNodeID, true } return masterNodeID, false } // releaseBuildNode 释放构建节点执行槽位 // Logan.Li func (h *PipelineHandler) releaseBuildNode(nodeID uint) { nodeMu.Lock() defer nodeMu.Unlock() if nodeRunningCount[nodeID] > 0 { nodeRunningCount[nodeID]-- } } // buildExecCtx 根据节点 ID 构建执行上下文 // Logan.Li func (h *PipelineHandler) buildExecCtx(nodeID uint, workspace string) (*PipelineExecCtx, error) { ctx := &PipelineExecCtx{ Workspace: workspace, NodeID: nodeID, NodeName: "Master", } if nodeID == masterNodeID { return ctx, nil } var node model.BuildNode if err := h.db.First(&node, nodeID).Error; err != nil { return nil, fmt.Errorf("构建节点不存在") } var srv model.Server if err := h.db.First(&srv, node.ServerID).Error; err != nil { return nil, fmt.Errorf("节点关联服务器不存在") } cfg, err := buildSSHConfig(h.db, srv) if err != nil { return nil, err } ctx.Remote = true ctx.SSHCfg = &cfg ctx.NodeName = node.Name return ctx, nil } // remoteWorkspacePath 计算远程工作区路径 // Logan.Li func remoteWorkspacePath(node model.BuildNode, repoName, code string) string { base := strings.TrimRight(node.WorkspacePath, "/") if base == "" { base = "/data/aiops-workspace" } return fmt.Sprintf("%s/%s-%s", base, repoName, code) } // execOnCtx 在 Master 或远程节点执行 shell 命令 // Logan.Li func (h *PipelineHandler) execOnCtx(ctx *PipelineExecCtx, workDir, command string, timeout time.Duration) (string, error) { if timeout <= 0 { timeout = 30 * time.Minute } if !ctx.Remote { cmd := exec.Command("bash", "-c", command) if workDir != "" { cmd.Dir = workDir } out, err := cmd.CombinedOutput() return string(out), err } fullCmd := command if workDir != "" { fullCmd = fmt.Sprintf("cd %s && %s", shellQuotePath(workDir), command) } result, err := remoteexec.Exec(*ctx.SSHCfg, fullCmd, timeout) output := "" if result != nil { output = result.Output } if err != nil || (result != nil && !result.Success) { if err != nil { return output, err } if result.Error != "" { return output, fmt.Errorf("%s", result.Error) } return output, fmt.Errorf("exit=%d", result.ExitCode) } return output, nil } // prepareWorkspace 初始化工作区(本地或远程) // Logan.Li func (h *PipelineHandler) prepareWorkspace(ctx *PipelineExecCtx, logFn func(string, string, string)) error { if !ctx.Remote { os.RemoveAll(ctx.Workspace) return os.MkdirAll(ctx.Workspace, 0755) } parent := filepath.Dir(ctx.Workspace) _, err := h.execOnCtx(ctx, "", fmt.Sprintf("rm -rf %s && mkdir -p %s", shellQuotePath(ctx.Workspace), shellQuotePath(parent)), 5*time.Minute) return err } // syncRemoteWorkspaceToLocal 将远程工作区同步到 Master(部署阶段需要本地产物) // Logan.Li func (h *PipelineHandler) syncRemoteWorkspaceToLocal(ctx *PipelineExecCtx, localWS string, logFn func(string, string, string)) error { if !ctx.Remote || ctx.SSHCfg == nil { return nil } logFn("info", "同步", fmt.Sprintf(">>> 从节点 %s 回拉工作区", ctx.NodeName)) tmpTar := localWS + ".tar.gz" os.RemoveAll(localWS) os.MkdirAll(localWS, 0755) if err := remoteexec.StreamRemoteTar(*ctx.SSHCfg, ctx.Workspace, tmpTar, 30*time.Minute); err != nil { return fmt.Errorf("远程打包下载失败: %v", err) } defer os.Remove(tmpTar) extract := exec.Command("tar", "xzf", tmpTar, "-C", localWS) if out, err := extract.CombinedOutput(); err != nil { return fmt.Errorf("解压工作区失败: %v, %s", err, string(out)) } logFn("success", "同步", "工作区已同步到 Master") return nil } // statOnCtx 检查远程/本地文件是否存在 // Logan.Li func (h *PipelineHandler) statOnCtx(ctx *PipelineExecCtx, path string) bool { if !ctx.Remote { _, err := os.Stat(path) return err == nil } out, err := h.execOnCtx(ctx, "", fmt.Sprintf("test -e %s && echo yes", shellQuotePath(path)), 30*time.Second) return err == nil && strings.Contains(out, "yes") } func shellQuotePath(s string) string { return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" } // pathExists 检查 Master/远程路径是否存在 // Logan.Li func (h *PipelineHandler) pathExists(ctx *PipelineExecCtx, path string) bool { return h.statOnCtx(ctx, path) } // hasBuildFileCtx 检查目录是否包含可识别的构建文件 // Logan.Li func (h *PipelineHandler) hasBuildFileCtx(ctx *PipelineExecCtx, dir string) bool { for _, f := range []string{"pom.xml", "package.json", "go.mod", "Makefile"} { if h.pathExists(ctx, dir+"/"+f) { return true } } return false }