Как мы используем Camunda в Банке ДОМ.РФ

Привет, Хабр! Наше подразделение занимается автоматизацией бизнес-процессов обслуживания клиентов в офисах и отделениях банка. Такие бизнес-процессы специфичны для конкретной организации, связаны с соблюдением внутренних регламентов и нормативных документов банка, должны полностью соответствовать часто меняющимся требованиям законодательства и всем инструкциям и правилам ЦБ РФ. Так, например, за процедурой открытия вклада стоит множество взаимодействий различных классов систем и компонентов, начиная от сбора и хранения согласий клиента на обработку персональных данных, интеграций с госсервисами и системами сбора сведений и заканчивая формированием приходно-кассовых ордеров. 

Подобные процессы мы реализуем с помощью движка Camunda, который позволяет прототипировать и визуализировать сложную последовательность шагов и моделировать flow совместно с бизнес-заказчиком, а также обладает множеством инструментов для реализации интеграций, оркестрации, мониторинга, сбора бизнес-метрик и тестирования. + быстро и относительно незатратно менять процессы (можно привести пример с запретом перевода валюты в другие банки). Вдобавок Camunda является open-source решением, что крайне важно в текущих реалиях и т.д. 

Впервые в наше подразделение Camunda попала в рамках закупки некоторого функционала у вендора. Реализация вендора была закрытой, сам движок инкапсулировался в каждый отдельный микросервис и был окружён различными самописными расширениями для централизованной разработки bpmn-процессов в браузере и сбора данных от всех запущенных инстансов. 

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

Архитектура 

Мы провели сравнение Camunda 8 и Camunda 7 по нашим внутренним критериям, проанализировали варианты развёртывания Camunda 7 (https://docs.camunda.org/manual/7.19/introduction/architecture/) и остановились на кластерной модели Camunda 7 с рекомендованной разработчиками Camunda архитектурой развёртывания c использованием удаленного движка, а точнее Camunda Run в качестве workflow-движка. 

Далее мы развернули отдельный инстанс СУБД Postgres только под задачи Camunda, несколько экземпляров Camunda Run (воркеры) в кластере за балансировщиком для исполнения процессов, а также отдельный независимый экземпляр Camunda Run исключительно для задач администрирования и мониторинга. 

Настройки Camunda Run для администрирования и мониторинга процессов: 

camunda.bpm:    job-execution:      enabled: false  

Настройки узлов Camunda Run (воркеров) за балансировщиком 

camunda.bpm:    job-execution:      max-wait: 10000      backoff-time-in-millis: 50      max-backoff: 250      backoff-decrease-threshold: 100      wait-increase-factor: 2 

 
Из коробки Camunda не умеет работать с Kafka, но большая часть наших процессов и внутренних систем базируются на event-driven паттерне, поэтому после первого этапа выбора архитектуры и развёртывания системы нам было необходимо подружить Camunda с Kafka: 

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

• для запроса действий в других системах в рамках исполнения различных бизнес-задач 

• для обеспечения интеграционных взаимодействий  

Укрупнённо план выглядел следующим образом: 

 

Где…  

Микросервисы активации процессов:

• являются держателями спроектированных bpmn-диаграмм 

• могут вести учёт и версионирование диаграмм 

• взаимодействуют с Kafka и умеют соотносить нужный процесс с тем или иным событием 

• обеспечивают идемпотентность на случай дублей или сбоев 

Микросервисы обработки внешних задач :

• используются вместо делегатов 

• обращаются к Camunda с заданной частотой для получения задач 

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

• могут запускать другие bpmn-процессы (через Kafka и взаимодействия с микросервисами активации процессов) 

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

Часть бизнес-процесса открытия вклада, завязанная на асинхронный обмен событиями по Kafka, после этапа прототипирования выглядела следующим образом: 

 

Здесь стоит сказать, что для взаимодействия с Kafka мы используем модель, предложенную Крисом Ричардсоном, 

Асинхронное API наших сервисов состоит из каналов и соответствующих типов сообщений, таких как: 

• входящие команды, 

• ответы на командные запросы и 

• события 

Схема асинхронного API сервиса из книги Ричардсона приведена ниже: 

 

Таким образом, возвращаясь к bpmn-схеме, кубик  «Отправка запроса открытия вклада» должен быть реализован отправкой командного события в топик Kafka. 

Внешняя задача для данного шага была реализована с использованием базового класса обработки задач, наследующего интерфейс библиотеки Camunda  

org.camunda.bpm.client.task.ExternalTaskHandler 

с преобразованием атрибутов объекта ExternalTask в расширенный под наши потребности DTO: 

/**   * Обобщенное выполнение задачи. *   * @param externalTask        внешняя задача   * @param externalTaskService сервис внешних задач   */  @Override  public void execute(@NonNull final ExternalTask externalTask,               @NonNull final ExternalTaskService externalTaskService) {   final TaskBO task = ExternalTaskToTaskBOMapper.EXTERNAL_TASK_TO_TASK_BO_MAPPER.map(externalTask); 

Для сквозной и межсистемной трассировки мы передаём в хедеры сообщения в Kafka корреляционный ключ инстанса работающего бизнес-процесса (correlationId), и этот же идентификатор вернётся нам в хедерах reply-события для связки ответа микросервиса открытия вклада с конкретным исполняющимся процессом.  

/**   * Отправка сообщения в топик    *   * @param task    DTO-представление внешней задачи   * @param message DTO для отправки в топик Kafka   */  public void sendCommand(@NonNull final TaskBO task,                                        @NonNull final Object message) {        kafkaProducer.sendCommand(task.getBusinessKey(), message, task.getHeaders());  } 

Кубик получения результата открытия депозита из топика Kafka также реализован внешней задачей. 

Мы реализовали подписку на заданный reply-топик микросервиса открытия депозитов и ожидали в нём появления ответного события по первоначальному командному запросу. 

String processCorrelationId =  externalTask.getVariable(CORRELATION_ID);  correlationIds.add(processCorrelationId); executor.execute(() -> {      try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>      (defaultConsumerProperties)) {               consumer.subscribe(Collections.singletonList(externalTask.getVariable(KAFKA_TOPIC)));            ...      }      ... 

Пока инстанс микросервиса обработки внешних задач был один – всё работало отлично, каждое входящее событие из reply-топика Kafka фильтровалось consumer-ом по бизнес-ключу и хедерам, и в случае совпадения значения correlationId в хедере с любым элементом внутреннего списка микросервиса correlationIds, задача успешно завершалась. Однако, как только мы пытались масштабироваться и увеличить количество инстансов микросервисов обработки задач – часть reply-событий для одного инстанса-подписчика попадали в обработку другого инстанса-подписчика, correlationId входящего события отсутствовал во внутреннем списке инстанса correlationIds, и исполнение внешних задач в процессе отваливалось по таймауту. 

Для исправления данной ситуации потребовалось переопределить свойства подписчика и помещать его в уникальную группу для того чтобы каждый инстанс микросервиса исполнения внешних задач получал каждое событие из нашего reply-топика Kafka: 

Properties consumerProperties = new Properties(defConsumerProperties);  consumerProperties.putAll(defConsumerProperties);  consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_TASK_GROUP_ID + UUID.randomUUID()); 

И далее применить данные свойства к consumer: 

executor.execute(() -> {      try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties))  {           String kafkaTopic = externalTask.getVariable(KAFKA_TOPIC);           consumer.subscribe(Collections.singletonList(kafkaTopic)); 

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

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

Первая итерация нагрузочного тестирования включала в себя один инстанс воркера (Camunda Run), 1 инстанс сервиса активации процесса и один инстанс сервиса обработки внешних задач и стартовые показатели ресурсов:

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

resources:    limits:      cpu: '1'      memory: 512Mi    requests:      cpu: '1'      memory: 512Mi 

Настройки деплоймента для микросервиса обработки активации процессов:

resources:    limits:      cpu: '1'      memory: 512Mi    requests:      cpu: '1'      memory: 512Mi  

Настройки деплоймента воркера:   

resources:    limits:      cpu: '4'      memory: 2G    requests:      cpu: '1'      memory: 512Mi 

В течение 30 минут на данной конструкции и ограничениях были запущены и успешно завершены 300 экземпляров одного и того же прототипа процесса открытия вклада, состоящего более чем из 50 отдельных блоков-шагов, без достижения экстремума. НТ продолжается.