Недавно передо мной встала задача — есть 250 тестов, каждый из которых занимает от 5 до 30 минут, а иногда и часы. Надо запустить их в параллель на одной машине, но не больше 16 одновременно. Ограничение связано с некоторым лимитированным ресурсом, а также количеством ядер CPU. Реальное число одновременно запущенных задач нужно вычислять динамически, то есть пойти простым путем и зарегистрировать имя или лэйбл ресурса заранее с помощью плагина Lockable Resource не получится.
В поисках решения я прошерстил интернет и нашел несколько вариантов, но… Они не понравились мне плохой утилизацией ресурсов или невыразительной визуализацией пайплайна. И я написал свой вариант, точнее, несколько. Ниже я расскажу об известных мне подходах, кратко опишу минусы и плюсы каждого, поделюсь своим решением и тем, как пришел к нему. Возможно, вы наткнетесь на этот текст, решая похожую задачу. Мне бы он помог.
Выполнение задач «группами» (и почему его можно отбросить сразу)
Первое, что приходит на ум и легко гуглится — разделить тесты на группы по N штук в каждой и запускать эти группы в цикле через parallel. Однако этот подход уместен только когда задачи выполняются за примерно одинаковое количество времени, чтобы ресурсы были постоянно загружены.
Возможно, код в статьи покажется кому-то не совсем Groovysh.
Я больше Python-программист и в основном использую циклы, а функции наподобие each или collect применяю только там, где тело простое и помещается в 1 строчку.
Код решения:
node {
List<Closure> groups = []
Integer groupsCount = 3
Integer taskCount = 12
(1..groupsCount).each { groups.add([:]) }
for (int i in 1..taskCount) {
groups[i % groupsCount][i] = runTest("${i}")
}
for (group in groups) {
parallel group
}
}
def runTest(name) {
return {
stage(name) {
println(name)
}
}
}В нашем же случае, напомню, тест может длиться от нескольких минут до нескольких часов. Решение не подходит.
Плюсы:
Легкая реализация
Не требует плагинов
Минусы:
Плохая утилизация ресурсов. Следующая группа тестов ждет, когда завершится последний тест из предыдущей группы. При этом чаще всего тесты выполняются за разное количество времени.
Ищем дальше!
Мой первый подход: скрипт на Python с использованием Queue и threading
Следующей в голову пришла мысль использовать Python-скрипт: создать с помощью threading нужное количество воркеров, а они будут брать задачи из очереди Queue. Помимо Python, подход потребует дополнительного репозитория со скриптом или дополнительной функции обёртки в Jenkins shared library, где можно разместить и сам скрипт.
Вот как выглядит код подобного решения через функцию в Jenkins shared library, которая выкачивает скрипт и запускает его с необходимыми параметрами:
runTasksInParallel.groovy:
def call(String testSuitePath, Integer threads) {
String scripDir = "${env.WORKSPACE}/runTestSuite"
String scriptName = "run_test_suite_in_parallel.py"
if(! fileExists("${scripDir}/${scriptName}")) {
sh(label: "create script directory", script: "mkdir -p ${scripDir}")
writeFile(file: "${scripDir}/${scriptName}", text: libraryResource(scriptName))
}
dir(scripDir) {
try {
String output = sh(
label: "running script via python",
script: """
python3 ${scriptName} \
--test-suite "${env.WORKSPACE}/${testSuitePath}" \
--threads ${threads} \
2>&1
""",
returnStdout: true
)
return output
} finally {
deleteDir()
}
}
}run_test_suite_in_parallel.py:
import logging
import argparse
import queue
import threading
from pathlib import Path
def run_test(q: object):
while True:
test_path = q.get()
try:
print(f"Running test from {test_path}")
except Exception as err:
logging.error(err, stack_info=True)
finally:
q.task_done()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Script for running tests in parallel")
parser.add_argument("--test-suite", dest="test_suite", action="store", required=True,
help="path to test suite directory")
parser.add_argument("--threads", dest="threads", action="store", required=True, type=int,
help="limit of concurrent threads")
args = parser.parse_args()
q = queue.Queue()
for _ in range(args.threads):
worker = threading.Thread(target=run_test,
args=(q),
daemon=True)
worker.start()
test_suite = list(Path(args.test_suite).glob("**/*.test"))
for test in test_suite:
q.put(test)
q.join()
Плюсы:
Легкая реализация, не требующая какой-то определённой версии Python или нестандартных модулей.
Хорошая утилизация ресурсов. Как только воркер заканчивает выполнять одну задачу, он сразу получает следующую.
Минусы:
Плохая визуализация. Всё скрыто внутри одного стэйджа. Прогресс выполнения задач никак не визуализируется, остается разве что читать вывод питоновского скрипта сплошным текстом.
Часть кода отделена от пайплайна и даже может находиться в других репозиториях.
Основное, что не устроило меня в этом подходе — визуализация. Хотелось видеть, какие задачи выполняются в текущий момент времени, какие прошли, а какие упали. Понимать, во сколько потоков выполняется пайплайн. И чтобы логи находились в разных стэйджах.
Я попробовал улучшить визуализацию, генерируя фэйковые стэйджи после выполнения скрипта. И это привело меня к решению без Python-скрипта, о котором я расскажу ниже. Но параллельно…
Использование семафора из плагина Concurrent Step (но оно сомнительно для прода)
Параллельно с этим я нашел плагин, который позволяет использовать семафоры в пайплайне. Решение заключается в следующем. Вначале мы создаем семафор с ограниченным количеством допусков. И запускаем через parallel наш список задач. А в каждой задаче проверяем семафор.
@groovy.transform.Field semaphore = null
node {
semaphore = createSemaphore(permit: 3)
Map tasks = [:]
Integer taskCount = 12
for (int i in 1..taskCount) {
tasks[i] = runTest("${i}")
}
parallel tasks
}
def runTest(name) {
return {
stage(name) {
acquireSemaphore (semaphore) {
println(name)
}
}
}
}Плюсы:
Легкая и понятная реализация. Семафоры присутствуют во многих языках и многим известны.
Хорошая утилизация ресурсов. Как только семафор разблокируется, сразу стартует следующая задача.
Минусы:
Сам плагин давно не обновлялся и есть вероятность, что его перестали поддерживать: в случае конфликта с другими плагинами в будущем он может не иметь обновления.
Также коллеги скинули комментарий от одного из мейнтейнеров Jenkins
Jesse Glick: From a brief glance at https://github.com/jenkinsci/concurrent-step-plugin I would say that it is designed incorrectly (confuses “native” Java threads with “virtual” CPS VM threads) and should not be used. Most or all of its steps probably could be reimplemented correctly while using the same Pipeline script interface.
Плохая визуализация. Все задачи как будто выполняются одновременно. Мы получаем длинный вертикальный столбик стэйджей в графе и не понимаем, во сколько потоков всё это выполняется и сколько ресурсов заняло.
Мой второй подход на основе опыта с Python-скриптом (и с замыканиями)
Генерируя фэйковые стэйджи для улучшения визуализации первого подхода, я понял, что с parallel можно попробовать тот же подход с очередью и воркерами, что я применял в Python-скрипте. Для этого помимо списка задач выполнения тестов, создадим ещё список воркеров, которые представляют из себя тоже задачи. Но их задачей будет не выполнение тестов, а вытаскивание из очереди задач с тестами и их запуск. Список воркеров как раз и будем передавать parallel. Для потокобезопасной работы со списком задач я решил использовать плагин Lockable Resource (он уже был у нас установлен), при этом никаких ресурсов заранее создавать не нужно.
Пример решения в виде функции в Jenkins shared library:
limitedParallel.groovy:
def call(Map params) {
Integer workerCount = params.workerCount ?: 1
String resourceName = params.resourceName ?: "cores"
List<Closure> queue = params.tasks ? params.tasks.clone() : []
Boolean failFast = params.failFast ?: false
Map workers = ["failFast": failFast]
for(int i in (0..workersCount-1).toList()) { // toList to make function serializable
workers["${resourceName}_${i}"] = getTaskFromQueue(queue)
}
parallel workers
}
def getTaskFromQueue(List<Closure> queue) {
return {
Closure task = null
while(true) {
lock("Thread safe ${env.BUILD_TAG}") { // Resource name should be unique among multiple concurrent job runs
task = queue ? queue.pop() : null
}
if (task) {
task()
} else {
break // queue is empty
}
}
}
}Плюсы:
Хорошая визуализация. Каждый стэйдж соответствует отдельной задаче. Наглядно видно, во сколько потоков выполняется пайплайн, какие задачи упали, какие прошли, и какие задачи идут в данный момент.
Хорошая утилизация ресурсов. Как только заканчивается одна задача, сразу же начинается следующая, и так пока очередь не иссякнет.
Минусы:
Если количество задач равно количеству потоков или меньше него, граф в плагине BlueOcean не отображает имена задачек, их место занимают имена потоков.
На этом подходе я и остановился. И хочу поделиться примером его использования.
Чтобы было видно, как тесты выполняются за разное количество времени, а разные потоки успевают вытащить из очереди разное количество задачек, я добавил sleep на время от 1 до 10 секунд.
node {
List<Closure> tasks = []
Integer taskCount = 12
for (int i in 1..taskCount) {
tasks.add(runTest("${i}"))
}
limitedParallel(tasks: tasks, workersCount: 4, resourceName: "core", failFast: true)
}
def runTest(String name) {
return {
stage(name) {
Integer t = Math.abs( new Random().nextInt() % (10 - 1) ) + 1
sh("sleep ${t}")
}
}
}Также эту идею можно реализовать через колбэк-функцию вместо заранее определенной очереди. Но мне больше нравится данное решение через очередь. Также для наглядности можно прикрутить счётчик, который будет показывать в поле "Описание" сколько задач прошло и сколько осталось.
p.s. Если вы видели или знаете другие варианты решения задачи при неизвестном заранее количестве ресурсов, делитесь в комментариях, это сделает материал полезнее!
