Jenkins PipelineのJenkinsfileでテスト並列実行を最適化する

 Jenkinsのpipelineを使うと、Jenkinsfileとしてジョブの定義・連携をGroovyで柔軟に記述できるようになります。そこでは例えば「変数を用いてノードやステージを横断する連携を実装する」「いろいろな外部スクリプトを実行して最適な実行条件を分析する」といった処理を容易に実装できます。

 今回はその具体例として、pytestのテスト実行時間の最適化を行うJenkinsfileを題材にします。

pytestのテスト並列実行の時間を最適化する

 対象は、pytestの2並列実行のジョブです。具体的には、Jenkins Pipelineで指定されたgitリポジトリから必要なファイルをチェックアウトし、その中のpytestのテストケースをノードサーバで実行して、結果をプッシュ& Jenkins Test Report出力する処理とします。実現にあたっては、(仮にテストケースが長大であるとして)ノードサーバを2つ用意し、テストを2分割してそれぞれのノードで並列実行させることで、テスト実行時間の短縮を狙います。

 このテストの2分割ですが、旧来の静的なジョブ定義では、テストケース数などで事前に静的分割せざるを得ない場合が多いです。一方でJenkins Pipelineを用いると、テスト実行時間が均等に分割されるように、テストケースを動的に2分割する処理を容易に実装できます。

Jenkinsfileの実装

 そのJenkinsfileの具体例を示します。ここでは以下を実装しています。

  • Prepareステージで、create_test_selection_file.py(実装例は後述)をシェル実行し、その標準出力をtest_selectionに格納します。
    • create_test_selection_file.pyは、テストケースファイルを、テスト実行時間が均等に2分割になるように2つにグルーピングし、結果をpytestの引数形式で標準出力するスクリプトです。過去のJUnit XML形式レポートからファイル名と実行時間のデータを抽出・集計することで実現します。
  • Testステージで、2つのノード:node1、node2にて、pytestのテストケースを並行実行します。pytestには、前ステージで得たtest_selectionを引数として指定します。
  • Reportステージで、pytestが生成したテストレポートファイルのプッシュと、Jenkins Test Reportへの出力を行います。

まとめると外に用意した集計スクリプトを呼び出して、ノードで実行するテストケースを動的に選択しています。

Jenkinsfileの実装例:

pipeline { 
    environment {
        test_selection = ""
    }
    agent any 
    stages {
        stage('Prepare') {
            steps {
                script {
                    def output = sh (
                        script: 'python create_test_selection_file.py',
                        returnStdout: true
                    ).trim()
                    test_selection = output.split(',')
                }
            }
        }
        stage('Test') { 
            steps {
                parallel 'automation test':{
                    node('node1') {
                        checkout scm
                        sh "pytest ${test_selection[0]} --junitxml=testresult/result1.xml"
                        sh "python push_test_report.py testresult"
                    }
                    node('node2') {
                        checkout scm
                        sh "pytest ${test_selection[1]} --junitxml=testresult/result2.xml"
                        sh "python push_test_report.py testresult"
                    }
                }
            }
        }
        stage('Report'){
            steps {
                git branch: 'main', url: 'git://hoge/integration-test.git'
            }
            post {
                always {
                    junit 'intagration_test/testresult/*.xml'
                }
            }
        }
    }
}
create_test_selection_file.pyの実装

 (主題ではないため簡易的なサンプルですが)例えば以下のようなものになります。

import glob
import xml.etree.ElementTree as ET
import numpy as np

NUMBER_OF_TESTNODE = 2

def select_test_by_time(test_files, resultfiles):
    test_result = {}
    all_time = 0.0
    for file in resultfiles:
        root = ET.parse(file).getroot()
        all_time += float(root.attrib['time'])
        for testcase in root.iter('testcase'):
            test_result[testcase.attrib['file']] = test_result.get(testcase.attrib['file'], 0) + float(testcase.attrib['time'])

    testsuite_time = 0.0
    node_no = 0
    output = ""
    for key, value in test_result.items():
        testsuite_time += value
        output += key + ' '
        if testsuite_time > all_time / NUMBER_OF_TESTNODE and node_no < NUMBER_OF_TESTNODE - 1:
            testsuite_time = 0
            output += ","
            node_no += 1
    output = output.replace('\\','/')
    print(output)

def select_test_by_number(test_files):
    test_group = list(np.array_split(test_files, NUMBER_OF_TESTNODE))
    output = ''
    for tests in test_group:
        output += ' '.join(tests) + ','
    output = output.replace('\\','/')
    print(output)

def create_test_selection_file():
    test_file = 'test/test*.py'
    input_file = 'testresult/*.xml'

    test_files = glob.glob(test_file)
    resultfiles = glob.glob(input_file)

    if resultfiles:
        select_test_by_time(test_files, resultfiles)
    else:
        select_test_by_number(glob.glob(test_file))

if __name__ == '__main__':
    create_test_selection_file()