Jenkins并行流水线：如何基于在线节点列表构建测试Stage？

现有代码 #

def generateTestStage(job_name, nodes_label) {
    return {
        stage(job_name) {
            node(nodes_label) {
                /* stage stuff */
            }
        }
    }
}

pipeline
for(item in nodesByLabel(label: "myNodes", offline: true)) { // nodesByLabel requires "Pipeline Utility Steps" plugin
    stagesMap[ "${item}" ] = generatePrepareStage("${item}")
}

stagesMap += JOB_MAP.collectEntries() { key, val ->
    [ "$key" : generateTestStage(key, val) ]
}

parallel parallelStageMap

需求与现存问题 #

需求：先准备标签myNodes下的所有可用节点，再执行测试任务，任务耗时极长（8节点并行需4小时），但节点可能中途离线或进入维护状态。

现存两种问题：

• 当nodesByLabel设置offline: false：任务启动前离线的节点，后续上线后不会触发generatePrepareStage，导致测试部署失败；
• 当nodesByLabel设置offline: true：任务会一直等待离线节点上线，可能导致任务耗时远超预期（甚至数天）。

解决方案思路 #

思路1：动态获取在线节点+准备阶段延迟执行 #

核心是不在 pipeline 初始化阶段一次性获取所有节点，而是在每个测试阶段动态检查节点状态，仅对当前在线的节点执行准备和测试逻辑，同时跳过离线节点（或加入重试机制）。

修改generateTestStage，在阶段内先获取当前在线节点列表，匹配到目标节点后再执行准备+测试：

def generateTestStage(job_name, target_label) {
    return {
        stage(job_name) {
            // 动态获取当前在线的目标标签节点
            def online_nodes = nodesByLabel(label: target_label, offline: false)
            if (online_nodes.isEmpty()) {
                echo "⚠️ 标签${target_label}下无在线节点，跳过当前阶段"
                return
            }
            // 选第一个可用节点（或根据需求选择特定节点）
            def target_node = online_nodes[0]
            node(target_node) {
                // 先执行准备逻辑（复用原generatePrepareStage逻辑）
                generatePrepareStage(target_node)()
                /* 执行测试逻辑 */
                echo "开始在节点${target_node}执行测试任务"
                // 测试代码...
            }
        }
    }
}

// 构建并行阶段时，直接基于JOB_MAP生成，无需提前获取节点
def parallelStageMap = JOB_MAP.collectEntries() { key, val ->
    [ key : generateTestStage(key, val) ]
}

parallel parallelStageMap

思路2：调整并行结构，分离节点准备与测试的容错逻辑 #

把节点准备和测试拆分为两个独立的并行阶段，给准备阶段添加超时限制，超时则标记节点为不可用，测试阶段自动跳过该节点：

// 生成带超时和重试的准备阶段
def generatePrepareStageWithCheck(node_name, timeout_minutes = 30) {
    return {
        stage("Prepare_${node_name}") {
            timeout(time: timeout_minutes, unit: 'MINUTES') {
                // 轮询检查节点是否在线，每5秒查一次
                waitUntil {
                    def node_obj = Jenkins.instance.getNode(node_name)
                    return node_obj != null && node_obj.toComputer().isOnline()
                }
                node(node_name) {
                    /* 准备逻辑：部署依赖、环境初始化等 */
                    echo "完成节点${node_name}的准备工作"
                }
            }
        } catch (TimeoutException e) {
            echo "⏰ 节点${node_name}超时未上线，跳过准备阶段"
            // 标记该节点为不可用，后续测试阶段跳过
            env["NODE_${node_name}_AVAILABLE"] = "false"
        }
    }
}

// 生成测试阶段，仅执行可用节点的测试
def generateTestStage(job_name, node_name) {
    return {
        stage(job_name) {
            if (env["NODE_${node_name}_AVAILABLE"] == "false") {
                echo "⚠️ 节点${node_name}不可用，跳过测试"
                return
            }
            node(node_name) {
                /* 测试逻辑 */
                echo "在节点${node_name}执行测试任务"
                // 测试代码...
            }
        }
    }
}

pipeline {
    agent none
    stages {
        stage("Prepare Nodes") {
            steps {
                script {
                    def prepareStages = [:]
                    // 获取所有带目标标签的节点（不管离线与否）
                    def all_nodes = nodesByLabel(label: "myNodes", offline: true)
                    all_nodes.each { node ->
                        prepareStages["Prepare_${node}"] = generatePrepareStageWithCheck(node)
                    }
                    parallel prepareStages
                }
            }
        }
        stage("Run Tests") {
            steps {
                script {
                    def testStages = JOB_MAP.collectEntries() { key, node_name ->
                        [ key : generateTestStage(key, node_name) ]
                    }
                    parallel testStages
                }
            }
        }
    }
}

关键优化点 #

• 动态节点检查：避免初始化阶段一次性获取节点列表，而是在执行阶段实时获取在线节点，适配节点状态变化；
• 超时容错：给离线节点的准备阶段设置超时，超时则跳过，避免任务无限等待；
• 状态传递：通过环境变量标记不可用节点，测试阶段自动跳过，保证整体任务能继续执行；
• 逻辑内聚：把准备逻辑嵌入测试阶段，或通过函数复用，避免节点状态变化导致的准备遗漏。

内容的提问来源于stack exchange，提问作者BouinA