Всем привет! Меня зовут Филипп Бороздин, я разработчик платформы GitFlic.
Вместе с командой мы создаем продукт, включающий в себя все возможности Git, автоматическое тестирование и анализ кода, CI/CD, реестр пакетов, а также множество других фич, с полным списком которых вы можете ознакомиться на сайте нашей платформы.

В данной статье я расскажу вам, как мы оптимизировали процесс загрузки артефактов CI/CD, используя чанковую загрузку.

Что такое артефакт?

Артефакт — это файл, полученный в результате выполнения задачи в конвейере CI/CD. Данные файлы сохраняются и могут быть использованы после завершения конвейера. Например, это могут быть скомпилированные бинарные файлы, необходимые для развертывания приложения или отчет о проведенных тестах, который позволяет анализировать качество кода.

Основные компоненты системы передачи артефактов

Передача артефактов в GitFlic происходит между двумя компонентами системы:

• GitFlic Runner (далее — Runner) — приложение, ответственное за выполнение задач CI/CD, формирование артефактов и их передачу.

• GitFlic Instance (далее — Instance) — центральный узел системы, обеспечивающий координацию задач CI/CD и управление состоянием Git репозиториев.

Почему возникла потребность оптимизировать процесс загрузки артефактов?

Изначально, при разработке процесса передачи артефактов CI/CD в GitFlic, мы использовали монолитную передачу файлов, отправляя их целиком. Этот подход удовлетворял нашим потребностям, пока размер артефактов оставался в пределах нескольких сотен мегабайт. Однако, когда пользователи начали передавать файлы большего размера, мы столкнулись с несколькими значительными проблемами:

1. Время передачи данных: передача файлов монолитным способом занимает большое количество времени, что значительно увеличивает время выполнения всего CI/CD процесса.

2. Неэффективность при сетевых сбоях: при возникновении ошибок во время передачи артефакта потребуется повторная отправка всего файла.

Решение

В связи с вышеперечисленным, мы были вынуждены искать альтернативные решения, которые позволят нам повысить эффективность и надежность процесса загрузки артефактов CI/CD. Для этого мы решили попробовать использовать принцип чанковой загрузки.

Как это работает?

После создания артефакта в Runner проверяется размер полученного файла. Если его можно разделить на несколько частей (чанков), применяется чанковая загрузка (о размере чанка читайте в блоке “Выбор размера чанка и количества потоков”)

Чанковая загрузка выполняется в несколько этапов:

1. Инициализация загрузки

2. Передача чанков

3. Завершение загрузки и проверка целостности данных

Прежде чем перейти к детальному разбору каждого из этапов, давайте рассмотрим SQL модель хранения данных передаваемых с помощью чанков:

Для реализации чанковой загрузки используются две таблицы: blob_upload и blob_upload_part.

Таблица blob_upload связывает части файла с общим процессом загрузки. Она хранит метаинформацию о передаваемом файле, пути для его сохранения, месте, где он будет храниться, времени начала передачи файла, а также уникальный идентификатор проекта, в рамках которого был запущен конвейер.

В свою очередь, таблица blob_upload_part служит хранилищем метаинформации о передаваемых чанках. Она хранит в себе hash_md5 части файла, его порядковый номер, размер чанка и внешний ключ, который связывает данную таблицу с blob_upload в отношении Many-To-One по атрибуту id.

Взаимодействие этих двух таблиц обеспечивает надежное управление чанковой загрузкой, позволяя контролировать процесс передачи файла на каждом из этапов. Это необходимо, так как при чанковой загрузке консистеность данных не обеспечивается до тех пор, пока все части файла не будут успешно загружены и не будет произведена проверка целостности данных (об этом процессе читайте в блоке “Завершение загрузки и проверка целостности данных”).

Этап 1: Инициализация чанковой загрузки.

Если артефакт можно разделить на части, Runner отправляет запрос на Instance для инициализации загрузки артефакта. Instance определяет путь для сохранения файла, место, где он будет храниться, а также уникальный идентификатор задачи, в рамках которого был запущен конвейер. На основе этих данных Instance создает новую запись в таблице blob_upload, которая, как было сказано ранее, связывает все части файла с процессом их загрузки.

Схема инициализации чанковой загрузки выглядит следующим образом:

Этап 2: Передача чанков

После инициализации загрузки Runner начинает процесс формирования чанков для их последующей загрузки на Instance. 

В процессе загрузки файл читается в цикле, в каждой итерации которого формируется чанк. Размер чанка зависит от количества байт файла, которые еще не были переданы. Если из их количества нельзя сформировать минимум два чанка, создается буфер равный размеру оставшихся байт, иначе, берется стандартный размер чанка (о размере чанка читайте в блоке “Выбор размера чанка и количества потоков”)

Вместе с каждой частью также передается ее порядковый номер и размер. Instance вычисляет hash_md5 полученной части и сохраняет всю метаинформацию в таблицу blob_upload_part, связывая новую запись с самим процессом загрузки, информация о которой хранится в таблице blob_upload.

Для увеличения скорости передачи данных каждый чанк отправляется в отдельном потоке, что позволяет загружать несколько частей файла параллельно.

Рассмотрим реализацию параллельной передачи чанков в коде:

Semaphore semaphore = new Semaphore(threadPoolLimit);

long bytesUploaded = 0;

int partNumber = 0;



try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, RandomAccessFileMode.READ_ONLY.toString())) {

   // Цикл, который продолжается, пока не будут загружены все байты файла

   while (bytesUploaded < file.length()) {

       final byte[] buffer; // Буфер для хранения данных текущей части

       long remainingBytes = file.length() - bytesUploaded; // Оставшиеся байты для загрузки


       partNumber++; // Инкрементируем номер передаваемой части файла


       // Проверка, чанк какого размера можем сформировать

       if (remainingBytes / (сhunkSize) < MIN_QUANTITY_CHUNKS_TO_UPLOAD) {

       // Если количество оставшихся байт файла меньше чем размер двух чанков

           buffer = new byte[(int) remainingBytes];

       } else {

       // Создается стандартный размер чанка

           buffer = new byte[(int) сhunkSize];

       }


       randomAccessFile.seek(bytesUploaded); // Двигаем курсор по файлу на количество уже прочитанных байт

       final int bytesRead = randomAccessFile.read(buffer); // Чтение данных из файла в буфер


       if (bytesRead == -1) {

           break; // Если достигнут конец файла, выходим из цикла

       }


       final long finalBytesUploaded = bytesUploaded; // Сохраняем текущее количество загруженных байтов

       final int finalPartNumber = partNumber; // Сохраняем номер части


       semaphore.acquireUninterruptibly(); // Ожидание освобождения семафора


       // Создание нового потока для загрузки текущего чанка

       new Thread(() -> {

           try {

               // Загрузка чанка

               uploadChunk(url, token, uuid, finalBytesUploaded, buffer, finalPartNumber);

           } finally {

               // Освобождение семафора после завершения загрузки

               semaphore.release();

           }

       }).start();


       bytesUploaded += bytesRead; // Увеличение количества загруженных байтов

   }


   // Ожидание завершения всех потоков перед выходом из метода

   semaphore.acquireUninterruptibly(threadPoolLimit);

   semaphore.release(threadPoolLimit);


} catch (IOException e) {

   throw new RuntimeException("Error uploading file in chunks: " + file.getName(), e);

}

Управление потоками

Для контроля количества одновременно активных потоков используются семафоры. Каждый раз, когда количество активных потоков достигает установленного предела threadPoolLimit (о данном ограничении читайте в блоке “Выбор размера чанка и количества потоков”), новые запросы ожидают освобождения ресурсов. Когда ресурс доступен, создается новый поток для загрузки текущего чанка.

В свою очередь, поля final необходимы для избежания проблем с конкурентным доступом между потоками.

В конце процесса передачи данных происходит ожидание завершения всех запущенных потоков перед выполнением завершения передачи чанков файла.

Процесс передачи чанков схематически будет выглядеть следующим образом:

Выбор размера чанка и количества потоков

Чтобы обеспечить быструю передачу файла, необходимо подобрать оптимальный размер чанка и количество потоков. Однако следует учитывать несколько факторов:

• В случае использования S3-хранилища для хранения артефактов минимальный размер одного чанка при multipart загрузке составляет 5 МБ.

• Большое количество потоков могут сильно нагружать сервер.

Проведя замеры скорости передачи файлов при различных условиях (размер файла, количество потоков, размер чанка), мы получили следующие результаты:

*Приведенные результаты тестирование содержат лишь часть полученных данных. Ознакомится со всеми данными можно по ссылке.
На основе полученных данных и проведенного анализа можем заключить, что использование размера чанка равным 10 МБ и 5 потоков даёт оптимальную производительность загрузки артефактов без лишней динамической логики.

Этап 3: Завершение загрузки и проверка целостности данных

После завершения передачи всех частей файла процесс загрузки чанков завершается. Чтобы убедиться в корректности передачи данных, проверяется их целостность.

Реализация проверки целостности данных

Для обеспечения надежности передачи данных мы используем сравнение вычисленных хэш сумм созданных чанков на стороне Runner и полученных чанков на стороне Instance. Вот как это происходит:

1. Вычисление хеша в Runner: при отправке чанка Runner вычисляет его MD5-хеш части и сохраняет внутри своей памяти.

2. Вычисление хеша в Instance: после получения чанка Instance также вычисляет ее MD5-хеш и сохраняется в таблицу blob_upload_part.

3. Сравнение хеш-сумм: после передачи всех чанков обе системы (Runner и Instance) суммируют полученные хеши, чтобы получить хеш-сумму всех частей. Если итоговые хеш-суммы на обеих сторонах совпадают, передача считается успешной.

Схема процесса проверки целостности переданных данных:

Что происходит с сохраненными ранее записями в таблицах blob_upload и blob_upload_part? 

В случае ошибок при передаче чанков (например, из-за перебоев в сети), удаление артефактов выполняется через worker — объект, работающий в отдельном потоке и выполняющий фоновые задачи. Этот worker каждые три часа очищает записи о начатых загрузках и информацию о полученных чанках, если они были созданы более трёх часов назад.

Если передача чанков произошла без ошибок, информация об инициализированной загрузке и успешно загруженных частях удаляется сразу после проверки целостности данных, вне зависимости от того с каким результатом закончилось выполнение данной проверки.

Схема полного жизненного цикла загрузки артефактов:

Подведение итогов

Для подтверждения эффективности выбранного нами решения были проведены тесты сравнения двух методов загрузки: монолитной передачи данных и чанковой загрузки. 

• Результаты тестирования чанковой загрузки 

• Результаты тестирования монолитной передачи данных:

После проведения сравнительного анализа мы обнаружили, что внедрение чанковой загрузки при оптимально выбранных параметрах размера чанка и количестве потоков (10мб и 5 потоков соответственно), уменьшило время передачи данных в 3,94 раза! 

Помимо этого, использование чанковой загрузки файлов эффективно решает проблему связанную с сетевыми сбоями. Если один из потоков прерывается или возникает ошибка во время передачи данных, можно перезапустить загрузку только этого чанка, не отправляя весь файл повторно.

Если интересно посмотреть, как это работает на практике - заходите на gitflic.ru, а если что-то покажется неочевидным или у вас есть идеи, как еще можно улучшить процессы разработки - пишите в чат сообщества или в комментарии.