feat: 工具配置增强、站点拨测与日志中心多项改进

- 工具配置：新增卸载 API/UI、Go 工具链、流水线构建自动安装依赖
- 修复 ListTools VerifyTool 递归栈溢出导致获取工具配置失败
- Node.js Alpine 安装改用 unofficial-builds musl 预编译包与 index.json 解析 LTS
- 站点拨测：详情页路由、告警自动恢复、并发拨测与默认可观测频率
- 日志中心：RBAC、查询构建器、ES/Loki 增强与审计 request/response 展示
- 发布配置：构建节点、流水线远程执行、Deploy RBAC 导航拆分

backend/controller/pipeline_executor.go

@@ -6,11 +6,13 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
 
 	"ezops/model"
+	"ezops/pkg/remoteexec"
 
 	"go.uber.org/zap"
 )
@@ -38,8 +40,9 @@ func init() {
 	logger, _ = zap.NewProduction()
 }
 
-// executePipeline 后台执行流水线（goroutine）— V2 Redis 版
-func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.PipelineRun) {
+// 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)
@@ -129,15 +132,47 @@ func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.Pip
 	}
 	ts := time.Now()
 	tsDir := fmt.Sprintf("%d%02d%02d%02d%02d", ts.Year(), ts.Month(), ts.Day(), ts.Hour(), ts.Minute())
-	workspace := fmt.Sprintf("/app/workspace/%s-%s", repoName, pipeline.Code)
+	localWorkspace := fmt.Sprintf("/app/workspace/%s-%s", repoName, pipeline.Code)
+	workspace := localWorkspace
 
-	os.RemoveAll(workspace)
-	os.MkdirAll(workspace, 0755)
-	// 保存 workspace 路径到运行记录，方便终端查看
-	h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Update("workspace_path", workspace)
+	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))
 
@@ -159,17 +194,31 @@ func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.Pip
 		var err error
 		switch stage.Type {
 		case "checkout":
-			err = h.runCheckout(gitURL, branch, workspace, log)
+			err = h.runCheckout(execCtx, gitURL, branch, log)
 		case "build":
-			err = h.runBuild(workspace, pipeline.Dockerfile, log)
+			err = h.runBuild(execCtx, pipeline.Dockerfile, log)
 		case "test":
-			err = h.runTest(workspace, log)
+			err = h.runTest(execCtx, log)
 		case "image":
-			err = h.runDockerBuild(workspace, pipeline, imageTag, log)
+			err = h.runDockerBuild(execCtx, pipeline, imageTag, log)
 		case "deploy":
-			err = h.runDeploy(workspace, pipeline, imageTag, log)
+			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":
-			err = h.runVerify(workspace, log)
+			verifyWS := execCtx.Workspace
+			if execCtx.Remote {
+				verifyWS = localWorkspace
+			}
+			err = h.runVerify(verifyWS, log)
 		default:
 			log("info", stage.Name, "跳过未知阶段类型: "+stage.Type)
 		}
@@ -214,25 +263,26 @@ func (h *PipelineHandler) flushLogsToDB(ctx context.Context, runID uint, logKey
 	h.cache.Del(ctx, logKey)
 }
 
-// runCheckout git clone
-func (h *PipelineHandler) runCheckout(url, branch, workspace string, logFn func(string, string, string)) error {
+// 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+")")
 
-	// U4: 确保工具可用（slim 镜像模式下 git 为懒加载）// Logan.Li
-	if h.toolInstaller != nil {
+	if !execCtx.Remote && h.toolInstaller != nil {
 		if err := h.toolInstaller.EnsureAvailable("git"); err != nil {
 			return fmt.Errorf("git 不可用（安装失败）: %v", err)
 		}
 	}
 
-	cmd := exec.Command("git", "clone", "--depth", "1", "-b", branch, url, workspace)
-	output, err := cmd.CombinedOutput()
-	// 流式输出到日志
-	for _, line := range strings.Split(strings.TrimSpace(string(output)), "\n") {
+	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)
 		}
@@ -240,19 +290,19 @@ func (h *PipelineHandler) runCheckout(url, branch, workspace string, logFn func(
 	if err != nil {
 		return fmt.Errorf("git clone 失败: %v", err)
 	}
-	hashCmd := exec.Command("git", "-C", workspace, "rev-parse", "--short", "HEAD")
-	hashOut, _ := hashCmd.Output()
-	logFn("success", "代码检出", fmt.Sprintf("已拉取 commit: %s", strings.TrimSpace(string(hashOut))))
+	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）
-func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(string, string, string)) error {
+// 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 _, err := os.Stat(testDir); err == nil {
+			if h.pathExists(execCtx, testDir) {
 				buildDir = testDir
 				logFn("info", "构建", ">>> 切换到构建目录: "+dir)
 			}
@@ -260,30 +310,48 @@ func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(stri
 	}
 
 	// 如果根目录没找到构建文件，扫描一级子目录
-	if !hasBuildFile(buildDir) {
+	if !h.hasBuildFileCtx(execCtx, buildDir) {
 		type subMatch struct {
 			dir  string
 			kind string // "maven", "npm", "go", "make"
 		}
 		var matches []subMatch
-		entries, err := os.ReadDir(buildDir)
-		if err == nil {
-			for _, e := range entries {
-				if !e.IsDir() || e.Name() == ".git" {
+		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
 				}
-				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"})
+				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"})
+					}
 				}
 			}
 		}
@@ -304,13 +372,19 @@ func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(stri
 		}
 	}
 
-	if _, err := os.Stat(buildDir + "/pom.xml"); err == 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", "构建", "🔨 正在下载依赖并编译...")
-		cmd := exec.Command("mvn", "clean", "package", "-DskipTests")
-		cmd.Dir = buildDir
-		out, err := cmd.CombinedOutput()
-		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+		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)
 			}
@@ -322,13 +396,16 @@ func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(stri
 		return nil
 	}
 
-	if _, err := os.Stat(buildDir + "/package.json"); err == 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", "构建", "📦 正在安装依赖...")
-		installCmd := exec.Command("sh", "-c", "npm install --registry=https://registry.npmmirror.com --legacy-peer-deps 2>&1")
-		installCmd.Dir = buildDir
-		out, err := installCmd.CombinedOutput()
-		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+		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)
 			}
@@ -337,10 +414,8 @@ func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(stri
 			return fmt.Errorf("npm install 失败: %v", err)
 		}
 		logFn("info", "构建", "🏗️ 正在打包构建...")
-		buildCmd := exec.Command("sh", "-c", "npm run build 2>&1")
-		buildCmd.Dir = buildDir
-		out, err = buildCmd.CombinedOutput()
-		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+		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)
 			}
@@ -352,12 +427,16 @@ func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(stri
 		return nil
 	}
 
-	if _, err := os.Stat(buildDir + "/go.mod"); err == 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 ./...)")
-		cmd := exec.Command("go", "build", "./...")
-		cmd.Dir = buildDir
-		out, err := cmd.CombinedOutput()
-		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+		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)
 			}
@@ -383,14 +462,12 @@ func hasBuildFile(dir string) bool {
 	return false
 }
 
-// runTest 执行测试
-func (h *PipelineHandler) runTest(workspace string, logFn func(string, string, string)) error {
-	if _, err := os.Stat(workspace + "/pom.xml"); err == nil {
+// 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)")
-		cmd := exec.Command("mvn", "test")
-		cmd.Dir = workspace
-		out, _ := cmd.CombinedOutput()
-		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+		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)
 			}
@@ -399,31 +476,15 @@ func (h *PipelineHandler) runTest(workspace string, logFn func(string, string, s
 	return nil
 }
 
-// runDockerBuild Docker 构建 & 推送
-func (h *PipelineHandler) runDockerBuild(workspace string, pipeline model.Pipeline, tag string, logFn func(string, string, string)) error {
+// 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"
 	}
 
-	var registryURL, imageName string
-	if pipeline.RegistryID != nil {
-		var reg model.PipelineRegistry
-		if err := h.db.First(&reg, *pipeline.RegistryID).Error; err == nil {
-			registryURL = reg.RegistryURL
-			if reg.Namespace != "" {
-				imageName = reg.Namespace + "/" + pipeline.Code
-			}
-		}
-	}
-	if imageName == "" {
-		imageName = pipeline.Code
-	}
-	if registryURL == "" {
-		registryURL = "registry.example.com"
-	}
-
-	fullImage := fmt.Sprintf("%s/%s:%s", registryURL, imageName, tag)
+	fullImage := h.resolvePipelineImage(pipeline, tag)
 	logFn("info", "镜像打包", fmt.Sprintf("镜像地址: %s", fullImage))
 
 	buildDir := workspace
@@ -437,17 +498,15 @@ func (h *PipelineHandler) runDockerBuild(workspace string, pipeline model.Pipeli
 
 	logFn("info", "镜像打包", fmt.Sprintf(">>> Docker 构建 (dockerfile: %s)", dfPath))
 
-	// U4: 确保 Docker CLI 可用（slim 镜像模式下 docker-cli 为懒加载）// Logan.Li
-	if h.toolInstaller != nil {
+	if !execCtx.Remote && h.toolInstaller != nil {
 		if err := h.toolInstaller.EnsureAvailable("docker-cli"); err != nil {
 			return fmt.Errorf("docker 不可用（安装失败）: %v", err)
 		}
 	}
 
-	cmd := exec.Command("docker", "build", "-t", fullImage, "-f", dfPath, ".")
-	cmd.Dir = buildDir
-	out, err := cmd.CombinedOutput()
-	for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
+	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)
 		}
@@ -458,9 +517,8 @@ func (h *PipelineHandler) runDockerBuild(workspace string, pipeline model.Pipeli
 	logFn("success", "镜像打包", "Docker 构建成功")
 
 	logFn("info", "镜像打包", ">>> 推送镜像: "+fullImage)
-	pushCmd := exec.Command("docker", "push", fullImage)
-	pushOut, err := pushCmd.CombinedOutput()
-	for _, line := range strings.Split(strings.TrimSpace(string(pushOut)), "\n") {
+	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)
 		}
@@ -472,8 +530,165 @@ func (h *PipelineHandler) runDockerBuild(workspace string, pipeline model.Pipeli
 	return nil
 }
 
-// runDeploy 部署到 K8s / 服务器
+// 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.RegistryID != nil {
+		var reg model.PipelineRegistry
+		if err := h.db.First(&reg, *pipeline.RegistryID).Error; err == nil {
+			registryURL = reg.RegistryURL
+			if reg.Namespace != "" {
+				imageName = reg.Namespace + "/"
+			}
+		}
+	}
+	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 {
@@ -485,12 +700,10 @@ func (h *PipelineHandler) runDeploy(workspace string, pipeline model.Pipeline, t
 			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)
@@ -501,14 +714,65 @@ func (h *PipelineHandler) runDeploy(workspace string, pipeline model.Pipeline, t
 			}
 		}
 		if err != nil {
-			logFn("error", "部署", fmt.Sprintf("kubectl apply 失败: %v", err))
-			return fmt.Errorf("kubectl apply 失败")
+			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", "验证", ">>> 部署后健康检查")
@@ -578,27 +842,51 @@ var (
 	queueMu              sync.Mutex
 )
 
-// tryExecuteOrQueue 触发流水线时检查并发限制
-// 如果当前运行数 < maxExecutors → 立即执行
-// 否则 → 加入队列 (status=queued)
+// tryExecuteOrQueue 触发流水线时检查流水线并发 + 构建节点槽位 // Logan.Li
 func (h *PipelineHandler) tryExecuteOrQueue(pipeline model.Pipeline, run model.PipelineRun) {
 	max := pipeline.MaxExecutors
-	if max <= 0 { max = 2 }
+	if max <= 0 {
+		max = 2
+	}
 
 	queueMu.Lock()
 	running := pipelineRunningCount[pipeline.ID]
-	if running < max {
-		pipelineRunningCount[pipeline.ID] = running + 1
+	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", "running")
-		go h.executePipeline(pipeline, run)
+		h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued")
 		return
 	}
-	// 加入队列
-	pipelineQueues[pipeline.ID] = append(pipelineQueues[pipeline.ID], run.ID)
 	queueMu.Unlock()
-	h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued")
+
+	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 流水线运行结束后，从队列中取下一个执行
@@ -625,8 +913,19 @@ func (h *PipelineHandler) dequeueNext(pipelineID uint) {
 	var run model.PipelineRun
 	if err := h.db.First(&run, nextRunID).Error; err != nil { return }
 
-	h.db.Model(&run).Update("status", "running")
-	go h.executePipeline(pipeline, run)
+	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 使用）