В Rusprofile наш домен — регтех, мы помогаем компаниям оценивать надежность контрагентов. Каждый день сотни скриптов качают данные из официальных открытых источников, парсят и строят кеши.

В какой‑то момент суммарное время работы скриптов стало неподобающим — задача заканчивалась, условно, к обеду следующего дня, в то время, когда нам нужно было к утру.

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

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

Будут технические детали, теория графов и немного рефлексии.

Проблема: зоопарк скриптов

Когда началась эта история, для запуска скриптов по расписанию мы использовали cron. Каждую ночь работал главный bash-скрипт, который запускал другие, дожидаясь их выполнения через wait. 

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

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

А ведь госзакупки — это только один из полусотни официальных открытых источников, откуда нужно получать информацию. В общей сложности за один обход всех источников отрабатывает 300+ скриптов.

Для управления всем этим хозяйством, чтобы работало за подобающее время, требовалось кардинально изменить подход.

Что там по готовым решениям? 

Мы пошли искать готовые решения. 

На дворе был 2018 год. Тогда уже существовал Airflow, но умел он значительно меньше, чем сейчас. Например, для нас была важна возможность динамически запускать задачи в зависимости от ресурсов, но Airflow в то время даже толком не гуглился под подобные запросы.

Что нагугливалось, так это сыроватые консольные утилиты, заточенные под узкие задачи, и академические статьи про работу с пайплайнами. Всё это нам не подходило.

Поэтому мы стали думать над собственным инструментом для управления скриптами.

Техника: проектируем инструмент 

Требования:

• унифицированный запуск разных скриптов;

• динамическая оценка используемых ресурсов;

• адаптивное планирование для запуска;

• поддержка сложных зависимостей между задачами;

• возможность изменения логики выполнения на лету;

• поддержка паузы и возобновления с места остановки.

У каждого скрипта есть цель — в духе «скачать новые файлы с FTP» или «распарсить новые XML».

Что важно для достижения цели:

• как именно запускается скрипт;

• сколько ресурсов занимает процесс;

• какие есть ограничения.

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

Обрабатываем задачи — запускаем скрипты. Например, для задачи, целью которой является «скачать файлы с FTP», будет запущен скрипт, скачивающий файлы и сохраняющий их в специальное хранилище. Информация о каждом запуске, успешном завершении, падении или пропуске — будем фиксировать в базе данных.

Цель TaskManager — сделать все задачи завершенными. 

Бэкенд мы пишем на PHP, в основу TaskManager заложили symfony/process: 

$process = Process::fromShellCommandline(‘'/bin/bash -e -o pipefail …команда’);
$process->start();

Здесь:

• “-e” заставляет bash завершить выполнение скрипта, если любая из простых команд возвращает ненулевой код выхода (произошла ошибка);

• “-o pipefail” возвращает код ответа для всего конвейера, если в одной из команд произошел сбой (например, если в скрипте есть конвейер в сочетании с xargs). 

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

if ($process->getExitCode() !== 0) {
  $stderr = $process->getErrorOutput();

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

Добавляем мониторинг: 

• в PHP-скрипты:

$data_pusher = $container->get(DataPusher::class);
$data_pusher->push([количество]));

• в bash-скрипты:

export SCRIPT_PID=$$

(сработает, даже если многопроцессность достигается через xargs);

• в любой процесс:

$data_pusher = $container->get(DataPusher::class);
$data_pusher->push(getenv("SCRIPT_PID"), [количество]);

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

Матчасть: ориентированные графы

Два ключевых вопроса, которые нам необходимо было решить при создании TaskManager: (1) планирование выполнения связанных между собой задач, (2) расход ресурсов.

Здесь нам на помощь пришли ориентированные графы.

Вершины графа — задачи.

Ребра — зависимости.

Визуализируем процесс:

Для отображения динамики процесса заводим цветовые обозначения:

Планирование выполнения связанных между собой задач

Допустим, у нас есть три скрипта:

• A — фетчер — скачивает файлы с FTP;

• B — парсер — проверяет скачанные файлы и раскладывает данные по таблицам;

• С — кешер — делает выборки на основе данных из таблиц, которые наполняет парсер.

В идеальном мире фетчер A может скачать файлы, парсер B — распарсить их, а скрипт построения кеша C — все посчитать:

Но что, если новые файлы появляются на FTP редко, так что фетчер A может не обнаружить их? Тогда парсер B не будет запущен, как и построение кеша C:

А что, если на FTP появился новый файл, но парсер не нашел в нем новых данных?

Или в процессе парсинга мы обнаружили некорректный формат данных и хотим прервать всю цепочку?

Даже в таком несложном примере вариантов протекания процессов может быть много. 

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

1. если процесс задачи уже запланирован или находится в работе или запускался нужное количество раз;

2. если процессы любой родительской задачи еще не отработали большее количество раз;

3. если процесс дочерней задачи (учитывается вся ветвь) еще в планировании или в работе или не запускался нужное количество раз.

Давайте посмотрим, как может выглядеть граф с процессами для примера, который мы уже видели выше:

1. планируются процессы задач A и B;

2. для С процесс не может быть запланирован — так как его задача зависит от A и от B;

3. D не может быть запланирован — так как зависит от B;

4. E не может быть запланирован — так как зависит от C;

5. запускаются запланированные процессы задач A и B;

6. по завершению A приносит новые данные, а B не приносит;

7. планируется процесс задачи C — так как один из зависимых процессов принес данные (процесс A);

8. D пропускается — так как зависит только от данных B, а данные не поступили;

9. E не может быть запланирован — так как зависит от C;

10. запускается запланированный процесс задачи C;

11. планируется процесс задачи B — так как в задаче задан один повтор;

12. планируется процесс задачи E — так как он зависит от C, который успешно отработал;

13. в процессе B произошла ошибка, а процесс E вып��лнился;

14. процесс D будет помечен упавшим с ошибкой — так как у него нет иммунитета к падению родительского процесса B, который упал с ошибкой.

Таким образом, реализация правил планирования процесса даёт нам следующую логику:  

После планирования проверяем, не отработали ли запущенные процессы, и если это так, то освобождаем ресурсы и запускаем новые процессы.

Теперь можно не задумываться над тем, чтобы в каждом скрипте прописывать условия ожидания, защиту от повторного запуска, реакцию на некоторые ошибки. Заодно убираем риски разрастания cron-файла или получения чрезмерно большого bash-сценария, в котором легко запутаться (не говоря уже о прописывании всех зависимостей вручную).

Расход ресурсов 

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

Рассмотрим пример: пусть наш вымышленный сервер обладает тремя ресурсами для обработки процессов. 

Посмотрим на уже знакомый нам граф:

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

Здесь видно, что ресурсы нашего сервера простаивают, и он способен обработать большее количество задач. Добавим еще один граф.

Теперь ресурсы используются более оптимально. 

И это не предел! Вот несколько идей по дальнейшей оптимизации ресурсов:

• используя знания о предыдущих запусках, можно разными способами прогнозировать время выполнения определенных процессов, тем самым более оптимально выбирать задачи для запуска из запланированных;

• расширить понятие «ресурсы» — например, завести отдельные характеристики по использованию ядер и оперативной памяти; прогнозировать потребление ресурсов на основе предыдущих запусков;

• можно добавить характеристику приоритетности (важности) задачи, с учётом особенностей продукта или предметной области;

• процессы могут занимать ноль ресурсов, если они обязательно должны запуститься — например, строго привязанных ко времени запуска.

Запас прочности решения 

Управление скриптами через TaskManager выглядит вот так:

Задачи и связи можно создавать двумя способами:

• через админку;

• в миграциях, используя готовые процедуры (разработчику в большинстве случаев приходится иметь дело именно с миграциями).

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

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

Работа скриптов и мониторинг стали наглядными. Есть возможность ретроспективно посмотреть историю за предыдущие дни.

Все это позволяет нам дирижировать сотнями связанных скриптов, многие из которых сами многопроцессные.

Стали бы мы делать TaskManager сейчас, если нужно было решать задачу по управлению скриптами в 2025? Не факт. Скорее всего попытались бы задействовать Airflow (с рядом костылей).

Сегодня мы используем Airflow, в пайплайнах аналитики. Справедливости ради, кажется, что ключевая ценность из коробки там всё же связана с удобным управлением периодичностью, в то время как TaskManager больше заточен под нашу специфику (нам всё ещё важно уметь динамически запускать задачи в зависимости от ресурсов).

Сколько ресурсов требует TaskManager в нашей инфраструктуре? Практически нисколько. Не нужны дополнительные сервера и базы данных, работает на уже существующих.

С учётом этих моментов можно попробовать, если не вычислить, то порассуждать о запасе прочности решения. Это как будто бы не абсолютная величина, в духе «инструменты такого‑то класса живут не больше N лет». Скорее производная — от прагматизма, готовности погружаться в проектирование (в том числе в алгоритмы) и стечения обстоятельств.

У тебя есть задача — придумай, как её решить.