Создание RPC-модулей

1. Описание

RPC (Remote Procedure Call) — это протокол взаимодействия между клиентом и сервером, позволяющий вызывать процедуры или функции на удалённом сервере так, как если бы они выполнялись локально. Этот подход используется при разработке распределённых и масштабируемых приложений, где отдельные компоненты взаимодействуют между собой, находясь на разных серверах или в разных сетевых средах.

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

Внутри инфраструктуры Cloudlink RPC-модуль подключается к заданной очереди RabbitMQ внутри Kubernetes-кластера. После разработки и и развертывания модуль можно интегрировать с Конструктором Cloudlink и использовать внешний сервис в качестве узла в графе.

Раздел содержит технические детали реализации RPC-модулей и предназначен для инженеров и разработчиков.

2. Предварительные условия

Для использования внешнего сервиса или приложения совместно с Конструктором Cloudlink нужно:

Разработать rpc-модуль для интеграции с Конструктором Cloudlink.
Добавить разработанный модуль в качестве шаблона узла в Конструктор Cloudlink.
Создать граф для продукта.

3. Разработка RPC-модуля

Развёртывание RPC-модуля может выполняться двумя способами:

Внутри Kubernetes-кластера Cloudlink:

RPC-модуль размещается в том же кластере Kubernetes, где работают все необходимые сервисы.
Вне Kubernetes-кластера:

RPC-модуль разворачивается за пределами кластера. В этом случае необходимо обеспечить корректное подключение к RabbitMQ, State Service и другим сервисам внутри кластера.

В качестве Service 1,2,3 может выступать любой компонент инфраструктуры Cloudlink.

4. Реализация RPC-модулей

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

Архитектура (высокоуровневый обзор)

Задачи поступают в RabbitMQ.
TaskReceiver читает сообщение и инициализирует Adapter.
Метод Adapter.execute(interface_schema) выполняет следующие действия:
- выбирает метод по полям _template или action_type;
- валидирует входные данные по соответствующей JSON Schema;
- вызывает бизнес-логику выбранного метода.
Схемы (JSON Schema) описывают входы конкретного метода и сопоставляются по названию: method ⇄ METHOD в interface_schema.

Архитектура (подробное описание)

Основные компоненты:

Adapter
- Базовый класс для адаптеров.
- Используется для инициализации, выбора и выполнения методов интерфейса.
- Используется для интеграции с внешними сервисами.
BaseTaskReceiver
- Класс для приема и обработки задач.
- Используется для управления асинхронным циклом и сигналами завершения, а также для валидации входящих сообщений.
TaskReceiver
- Расширяет BaseTaskReceiver, интегрируя пользовательские адаптеры.
- Использует методы адаптера для выполнения задач.

4.1. Контракт входного сообщения (`self.inputs`)

Источник сообщений: RabbitMQ → TaskReceiver.
Сообщение нормализуется в cld-state-service-utils и поле _type приводится к action_type.

В объекте self.inputs всегда присутствуют следующие поля:

_template: string — идентификатор интерфейса.
- Может иметь формат product:method.
- При remove_template_prefix=True используется часть после последнего :. Если : отсутствует, используется полное значение.
action_type: "run_node" | "rollback_node" — основной сценарий или сценарий отката.
Контекстные идентификаторы (могут использоваться для логирования и блокировок):
- _order_id (или order_action.order_id);
- _action_id;
- _graph_id;
- _id;
- _name.

Предметные поля — это пользовательские (ваши) параметры интерфейса, определяемые конкретным RPC-методом (например: param, product_name и т.п.). Они передаются через граф и описываются в JSON Schema.

Пример 1. Пример минимальной схемы

Служебные поля должны быть объявлены в properties, но не добавляться в required.

DO_SOMETHING = {
  "type": "object",
  "properties": {
    "action_type": {
      "type": "string",
      "enum": ["run_node", "rollback_node"]
    },
    "_template": {"type": "string"},
    "_order_id": {"type": "string"},
    "param": {"type": "string"},
  },
  "required": ["param"],
}

DO_SOMETHING_ROLLBACK = {**DO_SOMETHING}

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

4.2. Adapter (базовый класс)

Конструктор

Adapter(
  inputs: dict,
  redis_connection,
  logger,
  remove_template_prefix: bool = False
)

Параметры:

inputs — нормализованное входное сообщение.
logger — task-scoped логгер с контекстом (order_action, node, action_type).
- Инжектируется автоматически из TaskReceiver / EventsReceiver.
- Доступен как self.logger.
remove_template_prefix=True:
- из _template берётся часть после последнего :;
- если : отсутствует, используется полное значение.

Основные методы

execute(interface_schema) — точка входа выполнения задачи.

Метод последовательно выполняет:
- определение вызываемого метода на основе _template;
- валидацию входных данных по JSON Schema:
  jsonschema.validate(self.inputs, method_schema)
- вызов соответствующего бизнес-метода.
_obtain_interface_method() — вспомогательный метод выбора бизнес-метода:
- выбирает метод по значению _template;
- при action_type = "rollback_node" использует маппинг ROLLBACK_ACTIONS;
- как правило, не требует переопределения.
Если метод, соответствующий _template (с учётом ROLLBACK_ACTIONS), не найде, то выбрасывается ApiException с параметрами template_name и action_type.
_init_client() — опциональный метод для инициализации внешнего клиента. Он может возвращать экземпляр клиента или None.
interface_call() — стандартный способ вызова бизнес-метода:
- вызывает _init_client();
- при наличии клиента выполняет его в контекстном менеджере (async with);
- вызывает выбранный бизнес-метод.
  
  Метод interface_call() следует переопределять, если требуется обеспечить идемпотентность с помощью Redis-лока, работать с несколькими внешними клиентами, а также использовать клиентов, не поддерживающих контекстный менеджер.

Rollback

ROLLBACK_ACTIONS: dict[str, str] — маппинг основного метода на rollback‑метод.

Пример 2. Пример минимальной реализации адаптера

from cld_rpc_utils import Adapter

class MyAdapter(Adapter):
  # при откате будет вызван метод do_something_rollback
  ROLLBACK_ACTIONS = {"do_something": "do_something_rollback"}

  # реализовано в базовом адаптере, если нужен клиент
  # def _init_client(self):
  #   # return SomeClient(...) or None
  #   return None

  # Базовая реализация (как в библиотеке): клиент должен поддерживать async with, рекомендуется добавить redis lock
  # async def interface_call(self):
  #   self.client = self._init_client()
  #   async with self.client:
  #     return await self.interface_method()


  async def do_something(self):
    # read self.inputs[...] ; return dict
    return {"status": "success"}

  async def do_something_rollback(self):
    # логика отката
    return {"status": "success"}

4.3. TaskReceiver (входная точка)

from cld_rpc_utils import TaskReceiver
from app import schema, settings
from app.adapter import MyAdapter

# Инициализация логирования (вызвать один раз при старте)
# settings.setup_logging()

TaskReceiver(
  MyAdapter,
  app_name="rpc_myservice",
  task_execution_timeout=600,
  interface_schema=schema,
  redis_connection=True,
  # remove_template_prefix=True
).get_messages()

Параметры TaskReceiver:

plugin — класс адаптера.
app_name — название сервиса.
- Может приходить из узла графа или задаваться через ENV.
task_execution_timeout — таймаут выполнения одной задачи (в секундах).
interface_schema — модуль со схемами (константы в UPPER_CASE).
redis_connection — включить подключение к Redis (True / False).
remove_template_prefix=True:
- из _template берётся часть после последнего :;
- если : отсутствует, используется полное значение.

4.4. Процесс исполнения задачи с помощью библиотеки NovaAdapter

Получение сообщения из RabbitMQ

TaskReceiver прослушивает очередь RabbitMQ. При появлении нового сообщения (задачи) оно извлекается из очереди для дальнейшей обработки.
Обработка сообщения

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

TaskReceiver создаёт экземпляр адаптера (NovaAdapter) и передаёт ему входные данные и параметры логирования.
Выбор метода интерфейса

В NovaAdapter вызывается метод obtain_interface_method(), который определяет, какой метод интерфейса использовать на основе входных данных.
Выполнение задачи

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

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

Результат выполнения передаётся обратно в TaskReceiver. В зависимости от конфигурации, он может быть отправлен в очередь RabbitMQ или обработан другим способом.
Ожидание следующей задачи

После обработки текущего сообщения TaskReceiver продолжает прослушивание очереди RabbitMQ для получения новых задач.

5. Интеграция RPC со State Service

5.1. Описание работы

Библиотека cld_rpc_utils использует EventsReceiver из cld-state-service-utils, который автоматически:

Отправляет событие о начале задачи (STARTED) — перед выполнением вашего метода.
Нормализует _type → action_type — добавляет поле action_type в self.inputs для выбора метода. Также нормализаются _order_id → order_id, _action_id → action_id, _graph_id → graph_id для класса OrderAction.
Отправляет событие о завершении (COMPLETED) или ошибке (ERROR) — после выполнения или в случае исключения.

при STATE_SERVICE_MOCK=True декоратор сохраняет конвейер, но пропускает вызовы State Service — события не отправляются, что удобно для локальной разработки.

5.2. Обязательные переменные окружения

Для продакшн-окружения (без моков) обязательны: - STATE_SERVICE_URL — URL State Service API - KEYCLOAK_SERVER_URL, KEYCLOAK_REALM_NAME, KEYCLOAK_CLIENT_ID, KEYCLOAK_CLIENT_SECRET_KEY — для аутентификации

5.3. Ожидаемое поведение адаптера

Методы адаптера (ваши методы) должны возвращать структурированный результат в виде dict.
Результаты выполнения автоматически публикуются как события: прямые вызовы API State Service не требуются.
Необработанные исключения автоматически логируются и отправляются как события с типом ERROR.

5.4. ENV чек‑лист: универсально обязательные переменные

Сгруппировано по ключевым интеграциям. За деталями по State Service см. предыдущий раздел.

RabbitMQ (обязательно для всех RPC):

RMQ_HOST — хост брокера сообщений
RMQ_PORT — порт (обычно 5672)
RMQ_USER — имя пользователя
RMQ_PASSWORD — пароль
RMQ_COMMAND_QUEUE — название очереди для команд

Redis (обязательно, если используются локи)

REDIS_HOST — хост Redis
REDIS_PORT — порт
REDIS_DB — номер базы данных
REDIS_USER — имя пользователя
REDIS_PASSWORD — пароль
REDIS_LOCK_KEY — шаблон ключа для блокировок (формат: "myservice:order:%s")
REDIS_LOCK_TIMEOUT — таймаут блокировки в секундах (по умолчанию 3600)

State Service и Keycloak (обязательно для продакшн)

Без STATE_SERVICE_MOCK=True или DEBUG=True эти переменные обязательны, иначе модуль не запустится (EnvironmentError).

STATE_SERVICE_URL — URL State Service (например, https://api.domain.com/state-service)
KEYCLOAK_SERVER_URL — URL Keycloak (например, https://auth.domain.com/auth/)
KEYCLOAK_REALM_NAME — название realm в Keycloak (например, Portal)
KEYCLOAK_CLIENT_ID — ID клиента (например, cloud_rpc-myservice)
KEYCLOAK_CLIENT_SECRET_KEY — секретный ключ клиента

Для локальной разработки можно использовать:

STATE_SERVICE_MOCK=True — отключить отправку событий в State Service
DEBUG=True — режим отладки (не рекомендуется для продакшн)

References (обязательно в большинстве случаев)

REFERENCES_HOST_URL — URL справочников (например, https://keycloak.domain.com/references)
REFERENCES_MOCK — использовать моки вместо реального API (True для разработки, False для продакшн)

Логирование (рекомендуется явно указывать)

LOG_DIR — директория для логов (например, log)
LOG_DIR_PATH — альтернативное название для директории логов
LOG_FILENAME — название файла логов (например, rpc-myservice.log)
LOG_FORMAT — формат логов (например, JSON,CONSOLE или OPENSEARCH)
LOG_FILE_SIZE — размер файла в MB до ротации (например, 100)
LOG_FILES_COUNT — количество ротируемых файлов (например, 10)
LOG_LEVEL — уровень логирования (например, INFO)

6. Создание шаблона узла для RPC-модуля

RPC-модуль подключается к Конструктору Cloudlink через настроенный узел в Control Panel → раздел Шаблоны узлов. Этот узел обеспечивает связь с необходимой очередью RabbitMQ для обработки запросов.

Подробнее о настройках Шаблонов узлов в Руководстве администратора Cloudlink.

Модуль можно развернуть как внутри Kubernetes, так и в другой среде. При этом важно обеспечить доступ к:

очереди RabbitMQ
State Service
остальным сервисам, с которыми модуль должен взаимодействовать.

Порядок действий

Перейдите в раздел Control Panel Конструктор → Шаблоны узлов и нажмите +.

Во вкладке Основное укажите следующие параметры:

Наименование
Код шаблона
Описание
Название очереди для старта задачи
Название очереди для отката

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

Тип

Названия очереди для старта и отката задачи должны быть согласованы между разработчиком RPC-модуля и разработчиком продукта или графа. Для предварительного создания очереди, необходимо подключиться к RabbitMQ по ссылке https://rabbitmq.FQDN_Cloudlink/ и выполнить следующие действия:

Перейдите на вкладку Queues и найти пункт Add a new queue.
В качестве параметра Virtual host выберите корневой каталог /.
Выберите значение параметра Type из следующих вариантов:
- Default for virtual host
- Classic
- Quorum
- Stream
Введите название в поле Name. Оно должно совпадать с названием очереди в шаблоне узла.
Укажите длительность сохранения очереди в параметре Durability.
При необходимости указать дополнительные параметры в поле Arguments.

Во вкладке Параметры укажите входные и выходные параметры, которые нужны для работы вашего RPC-модуля.

Указанные параметры передаются в виде сообщения при обмене данными между RPC-модулем и приложением.

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

Созданный шаблон узла вы можете использовать в конструкторе графа.

7. Использование RPC-модуля в качестве узла в графе

Порядок действий

Перейдите в раздел Конструктор → Графы.
Нажмите кнопку +.
В вкладке Общая информация введите следующие параметры:
- Код графа — укажите уникальный символьный идентификатор продукта. Он может содержать только прописные латинские символы, цифры, нижнее подчеркивание,тире двоеточие и точку.
- Наименование — дайте уникальное название создаваемому графу.
  
  Код графа,наименование и автор необходимы для идентификации и управления графом в системе.
- Тип — укажите один из двух типов acting или creating.
- Описание(опционально) — опишите отличительные особенности продукта.
- Автор — укажите автора продукта.
- При необходимости активируйте опции, которые включаются в случае некорректного проигрывания графа:
  - Переводить заказ в статус "Ошибка".
  - Блокировать заказ при ошибке.
Добавьте узлы графа:
1. Перейдите во вкладку Узлы и нажмите кнопку +.
2. На вкладке Основное укажите следующие параметры:
  - Название
  - Описание
  - Выберите Шаблон узла с RPC-модуля.
  - Перейдите на вкладку Параметры и введите данные:
    
    В блок кода Static data.
    
    В блоках кода Input и Output будут указаны входные и выходные переменные, которые были ранее указаны при создании Шаблона узла. При необходимости их можно отредактировать.
    
    Данные, указанные в Static data можно использовать во всем графе. *
  - На вкладке Дополнительное заполните параметры, касающиеся запуска узла.
Нажмите Добавить, чтобы завершить создание узла.

Узел будет добавлен в граф.
Перейдите на вкладку графа Параметры заказа и задайте параметры:
- В блоке Форма заказа опишите структуру в JSON-формате для определения структуры данных и параметров заказа:
  Пример 3. Описание параметров в структуре JSON
  Здесь создаётся новый объект с наименованием New app. Для него в форме заказа должны быть заданы следующие поля: vars, credentials, net_segment.
  
  Поля vars и credentials имеют строковый тип.
  
  Значения задаются по шаблону (pattern): "^[a-zA-Z0-9-]*$ (латинские буквы в верхнем и нижнем регистре, цифры от 0 до 9, символы - и ).
  
  Ограничения по длине:
  
  минимальное количество символов — 3 ("minLength": 3),
  
  максимальное количество символов — 64 ("maxLength": 64).
  
  Параметр net_segment имеет тип object и позволяет выбирать значения из списка.
  
  Для создания пользовательского интерфейса формы заказа используется UI-схема.
  UI схемы позволяют визуально организовать поля ввода, кнопки и другие элементы интерфейса, обеспечивая интуитивно понятное и удобное взаимодействие для пользователя.
  
  Пример 4. Переопределение виджетов и полей в интерфейсе UI-схемы
  
  Для полей vars и credentials свойство ui:widget задаёт тип отображения элемента формы. В данном случае используется "ProductLabelWidget", который отображает поле как текстовую метку с названием продукта.
  
  Для поля net_segment свойство ui:field определяет интерфейс для работы со связанными объектами. Значение "DirectoryUiListField" в пользовательском интерфейсе будет выглядеть как поле со списком значений с доступными шаблонами на платформе.
- Перейдите на вкладку Детализация цены и выберите продукт для расчета стоимости и получите детализацию заказа.
Во вкладке Модификаторы выберите тип среды для исполнения графа:
- dev
- prod
- test
Нажмите Сохранить, чтобы завершить создание графа.