Flow Manager

Назначение

Сервис создания, хранения, версионирования и исполнения процессных пайплайнов (flows). Оркестрация пайплайнов с поддержкой Starlark-скриптов, in-memory тестового запуска и real-time стриминга событий через NATS.

Порты

• HTTP: 8098
• БД: flow_manager_db

Ключевые возможности

• CRUD пайплайнов с контролем доступа по владельцу
• Версионированные шаблоны — каждое сохранение создаёт иммутабельную версию
• Модель графа: DAG из нод, связанных рёбрами (edges)
• In-memory тестовый запуск (без записи в БД) с потоковой передачей событий через NATS
• Production-запуск с сохранением в БД и отслеживанием статуса
• Тестирование отдельных нод (code, job) через /try/...
• Ограничение параллельных тестовых запусков: максимум 50 одновременных
• Starlark-исполнитель: print(), setVar(), getVar(), fetch(), setupProcess(), runProcess(), общий ctx
• Артефакты: глобальный модуль artefacts — загрузка, извлечение, создание ZIP-архивов, сохранение в песочницу
• Файловая система: глобальный модуль files — чтение, запись, удаление файлов во временном каталоге, экспорт в песочницу
• Двойная аутентификация: JWT (пользователи) и Service Token (сервисы)
• Swagger UI

Статусы инстансов

Статус	Описание
idle	Создан, ожидает запуска
pending	Подготовка к запуску
running	Выполняется
completed	Успешно завершён
failed	Завершён с ошибкой
cancelled	Отменён

Типы нод

Control

Нода	Описание
startNode	Точка входа пайплайна. Не выполняет никакой логики, только помечает начало выполнения.

Input

Нода	Описание
inputButtonNode	Вход по нажатию кнопки. В тестовом режиме — pass-through (выполняется без логики).
inputEventNode	Вход по событию (NATS). В тестовом режиме — pass-through.
inputFormNode	Вход по отправке формы. В тестовом режиме — pass-through.
inputHttpRequestNode	Вход по HTTP-запросу. В тестовом режиме — pass-through.

Transform

Нода	Описание
transformCodeNode	Выполняет Starlark-скрипт (Python-подобный язык) над общим контекстом. Доступные функции описаны ниже в разделе transformCodeNode.
transformSetVariableNode	Устанавливает одну или несколько переменных в контексте из JSON-источника (data.source).
transformExtractArtifactNode	Извлекает артефакты из выполненного jobNode. В тестовом режиме — pass-through.
transformMergeProcessNode	Синхронизация: ожидает завершения всех параллельных веток runProcess перед продолжением.

Logic

Нода	Описание
logicConditionNode	Вычисляет Starlark-выражения как условия и направляет выполнение на соответствующий выходной порт (input-N). Если ни одно условие не выполнено — переход на порт error.
logicSwitchNode	Узел маршрутизации/переключения. В тестовом режиме — pass-through.

Output

Нода	Описание
outputHttpNode	HTTP-выход (отправка ответа). В тестовом режиме — pass-through.
outputEmailNode	Email-выход (отправка письма). В тестовом режиме — pass-through.

Job

Нода	Описание
jobNode	Создаёт и мониторит задачу через API command-manager. Ожидает завершения (таймаут 5 минут, интервал опроса 2 сек). Собирает артефакты по завершении.

---

Контекст выполнения

Каждый пайплайн имеет общий контекст (execution context) — хранилище переменных, передаваемое между нодами. При переходе к следующей ноде контекст клонируется, что позволяет ветвление с изолированными состояниями.

Автоматические переменные

Переменная	Описание
artefacts.previous	Список ID артефактов от последнего jobNode
artefacts.all	Все артефакты, накопленные с начала пайплайна

Рабочая директория

Каждый инстанс пайплайна получает временную директорию ARTIFACTS_TMP_DIR/<instance_id>. Директория удаляется после выполнения ноды. Модули files и artefacts работают только внутри этой директории.

---

transformCodeNode

Нода выполняет Starlark-скрипт (Python-подобный sandboxed-язык) в контексте текущего выполнения пайплайна. Скрипт имеет доступ к общему контексту (ctx), глобальным модулям (files, artefacts) и набору встроенных функций.

Полное описание всех функций, контекста и хелперов — в Starlark API.

WebSocket Events

Flow Manager публикует события в NATS. Gateway подписывается на эти subjects и транслирует их клиентам через WebSocket.

• Тестовые запуски: pipeline.test.<pipe_uuid>.node.*
• Production Run: pipeline.run.<instance_id>.node.*
• Платформенные: job.created, job.changed, executor.status, runner.change

Полное описание всех событий с пейлоадами — в WebSocket Events.

Конфигурация

Переменная	По умолчанию	Описание
PORT	8098	Порт сервера
HOST	0.0.0.0	Адрес сервера
GIN_MODE	debug	Режим Gin (debug/release)
DB_HOST	localhost	Адрес PostgreSQL
DB_PORT	5432	Порт PostgreSQL
DB_USER	flow_manager_user	Пользователь PostgreSQL
DB_PASSWORD	—	Пароль PostgreSQL
DB_NAME	flow_manager_db	Имя базы данных
DB_SSL_MODE	disable	Режим SSL
JWT_SECRET	—	Секрет для JWT
JWT_ALGORITHM	HS256	Алгоритм JWT
JWT_TTL	24h	Время жизни JWT
SERVICE_TOKEN	—	Токен межсервисного взаимодействия
MESSAGING_URL	nats://localhost:4222	URL NATS
ARTIFACTS_TMP_DIR	./tmp/flow-manager	Каталог для временных файлов артефактов
SANDBOX_DIR	./sandbox	Постоянный каталог для files.export() и artefacts.save() (структура: {SANDBOX_DIR}/{pipeline_id}/)
TEST_PIPELINE_TTL	10m	Время жизни тестового пайплайна в памяти
LOG_LEVEL	info	Уровень логирования
LOG_FORMAT	json	Формат логов