Гарантированная доставка интеграционных сообщений

Каждая малая компания или стартап хочет роста и развития в некотором будущем. Вслед за кропотливой работой сотрудников, выстраивании репутации компании приходит долгожданный успех. И вот технологии, которые были достаточны для решения задач первое время, перестают удовлетворять потребности пользователя. В компании появляются ИТ-ландшафт, различные ИТ-решения. За ростом и развитием компании будут приходить не только увеличение ее финансовых активов и числа сотрудников в штате, но и существенное изменение нагрузок на ИТ-окружение. Информация, собранная по крупицам, уходит на хранение в базы данных и однажды наступает момент, когда эту информацию необходимо синхронизировать или актуализировать в режиме реального времени. Здесь мы уже говорим о миллионах данных, которые большими пачками транслируются в систему-потребитель.
При внедрении и проектировании архитектуры решения важно помнить и учитывать, что помимо happy flow (все системы доступны, данные корректны, нет сбоев в ИТ-ландшафте) всегда есть пути обработки ошибок. А вот ошибок, которые могут возникнуть в сложно организованном пространстве - бессчетное множество.
Итак, нашей задачей была синхронизация объектов 365 и некоторой Мастер Системы. Под Мастером, конкретно в этом кейсе, мы понимаем не систему с эталонными и неизменными данными, а систему, в которой должна быть абсолютно вся актуальная информация по объекту в каждый момент времени. Оба Программных продукта находились в контуре компании (365 - версия onPremises).
Для синхронизации данных "365 → Мастер" была предусмотрена следующая схема: При изменении объекта в 365 срабатывает триггер, который отправляет запрос в Микросервис. Далее Микросервис обрабатывает полученное сообщение, преобразовывает в нужный вид и передает информацию в очередь (у Мастера была своя Kafka) системы потребителя. Мастер самостоятельно вычитывал сообщения очереди Kafka и выполнял необходимые действия с данными.


В обратном направлении ("Мастер → 365") синхронизация выглядела другим образом: При изменении объекта в Мастере, Мастер кладет сообщение об изменении в очередь в Kafka. Далее Микросервис вычитывает сообщение из Kafka, преобразовывает данные для вызова стандартного метода API 365 по обновлению объекта, вызывает стандартный метод API 365, затем, также по API, запускает Бизнес-процесс.


Мы заложили, что в случае возникновения ошибки недоступности сервиса при интеграционном вызове на периферии 365 необходимо повторить попытку вызова трижды. Т.е. если получен ответ 500 (и ему подобные) происходит троекратное повторение запроса через несколько секунд. Далее попытки вызова прекращаются и Администратор ставится в известность о возникновении проблемы. С ошибкой 400 (и ей подобных) никаких попыток не предполагалось - сразу высылалось уведомление Администратору.
В целом схема оказалась рабочей и даже отказоустойчивой при увеличении частоты поступлений запросов на обновление объекта (RPS). НО! т.к. 365 это не просто рядом стоящая система, а система, вписанная в ИТ-ландшафт, она была зависима от выделенных ресурсов, корректности настроек сети и многих других вещей по DevOps части. Здесь и начались проблемы... Произошел инфраструктурный сбой. Причем он произошел так, что взаимодействие Мастер-системы и Kafka работало корректно (там обеспечена сетевая связность), очередь сообщений Kafka вычитывалась, а вот стык 365 и Микросервиса давал сбой. Конечно, без решения корневой проблемы бессмысленно восстанавливать передачу данных, но вот последствия инцидента решать необходимо и задача быстрого и качественного восстановления синхронности информации становится лейтмотивом ближайших дней.
Интересно, что в случае с Мастер системой и Kafka все достаточно просто: есть набор записей, который обязательно доберется до очереди, зафиксируется в этой очереди в хронологическом порядке и дальше просто ждет вычитки и записи в объекты в системе. У 365 такого механизма нет, соответственно, никакой гарантии доставки интеграционного сообщения, историчности поступления запросов на обновление информации тоже нет. Итого при обновлении данных Мастер → 365 или 365→Мастер у нас всегда есть риск потери интеграционных запросов и хранения данных в неактуальном формате. Да, три попытки прошли, уведомление Администратору отправили, но Админ все-равно не знает, какой срез данных является целевым.
Кажется, что решение этой проблемы на поверхности - будем отправлять интеграционные запросы до тех пор, пока не придет успешный ответ. Но в таком случае мы можем попасть в ситуацию скопления большого количества интеграционных запросов и, как итог, падение системы.
Вторым и по итогу выбранным вариантом решения стала настройка очереди сообщений, аналогично Мастеру. Т.е. к 365 в том же PostgreSQL, без потери сетевой связности создали таблицы для записи интеграционных запросов. Структура таблицы достаточно проста:

object_id, ID объекта (тип данных: uuid) - Копируется из элемента приложения через SQL запрос создания задания в очереди;
body, JSON со слепком объекта в момент создания задания (тип данных: Jsonb) - Копируется из элемента приложения через SQL запрос создания задания в очереди;
record_id, GUID сообщения в очереди, который будет использоваться как идентификатор в сквозном процессе (тип данных: uuid) - Заполняется автоматически при добавлении записи в таблицу;
date_time, Дата/время добавления задания в очередь (тип данных: timestampz) -Заполняется автоматически при добавлении записи в таблицу body;
date_time_update, Дата/время последнего обновления задания (тип данных: timestampz) - Обновляется из ms;
description, Если при обработке задания в ms возникла ошибка, то заполняется текстом ошибки (тип данных: text) - Обновляется из ms;
request_type, Код объекта для отслеживания (тип данных: text) - Заполнение фиксированного значения через SQL запрос создания задания в очереди;
request_state, Статус обработки задания (1- создан, 2 - получен ms, 3 - Успешно обработан, 4 - обработан с ошибкой) (тип данных: int) - Заполняется при создании задания устанавливается в значение 1. В дальнейшем обновляется из ms;
counter, Числовая уникальная последовательность (тип данных: int8) - Заполняется автоматически при добавлении записи в таблицу.​

Микросервис выполнял запись в таблицу в PG, и только после этого происходила вычитка и ставились задачи на обновление и запуск процессов в 365. По введенной статусной модели запросов можно ориентироваться в интеграционных потоках - какие запросы дошли, а какие "застряли" в интеграционном слое и ожидают отбработки, а также восстановить поток информации в хронологическом порядке. Три попытки на отправку запроса МС в 365 мы также оставили, и если происходит что-то непредвиденное - уведомляем Администратора.

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

У вас микросервис по сути дублировал очередь кафки, но в таблицу в PG, верно?
Спасибо, что поделись опытом!

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