Example

---

Общая структура

• admin -- приложение-админка.

  • auth -- сервис авторизации.
  • cfm -- сервис отправки OTP-кодов подтверждений.
  • front -- см. admin/front/README.md.
  • gateway -- api-шлюз для BE, точка входа.

• infrastructure -- инфраструктурные элементы, вынесенные отдельно для упрощения.

  • kafka
  • mysql
  • redis

• monitoring -- приложение для мониторинга системы.

• node -- обвязки физического сервера, на котором развернут проект. При физическом разделении компонентов эти обвязки были бы частью каждого отдельно развернутого компонента.

Схема:

---

Структура компонента

Структура компонента нижнего уровня:

|- app          -- код приложения, конфиги, ресурсы, ...
|- build        -- элементы сборки контейнеров (Dockerfile'ы, конфиги, ...).
|- state        -- состояние развернутого приложения, сохраняемое между запусками (логи, данные, ...).

app

Структура основана на Hexagonal Architecture.

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

Используемые слои:

• build -- слой окружения и сборки.

• adapt -- не является полноценным слоем, содержит secondary-adapter`ы. Выделен из классического слоя инфраструктуры для наглядности.

• api -- клиенты ядра приложения (например, http-сервер, планировщик, ...), primary-adapter`ы.

• opera -- классический операционный слой.

• domain -- классический доменный слой.

• data -- альтернатива комбинации opera и domain для случаев, ориентированных на прямую работу с данными.

• infra -- классические инфраструктурные элементы, за исключением secondary-adapter`ов.

• std -- утилитарный код, который можно рассматривать в рамках проекта как часть стандартной библиотеки языка. Например, тип данных Email, функции работы с массивами, ...

Направление зависимостей -- строго сверху вниз: вышележащий слой имеет доступ только к примыкающим снизу слоям (см. схему).

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

TODO: ??? обращение к adapt из api

gateway/session_closed

Доступ к смежным слоям выполняется через интерфейсы и адаптеры.

В зависимости от приложения те или иные слои могут отсутствовать.

client не входит в приложение, но изображен на схеме для наглядности.

build

TODO

state

TODO

---

Запуск

1. Копировать .env.example в .env, переопределить переменные при необходимости.

2. Выполнить

   . tools.sh up
   

---

Заметки

Docker-compose и .env

Файлы не распределены по папкам компонентов (для пущей инкапсуляции), поскольку могут использовать общие элементы (например, DEPLOY_... переменные окружения). Также такой подход усложнил бы использование переменных окружения и запуск проекта (например, при конфликте переменных API_KEY в файлах /cmp1/.env и /cmp2/.env).

Сборщики логов и ротация.

Для экспорта логов в систему мониторинга используются grafana-alloy-контейнеры. Для ротации логов на сервере используется logrotate.

При текущей конфигурации (alloy-экспортеры c интервалами сбора 5s и logrotate в режиме copy-truncate) возможна ситуация, когда между сборами логов alloy-экспортером произойдет запись новых логов и ротация файла. Тогда логи, записанные в промежуток между последним сбором и ротацией, не будут собраны, поскольку уже "архивированы".

Кроме того, logrotate с copy-truncate сам по себе может терять логи во время выполнения (см. в конфиге logrotate).

Включение "архивных" файлов в сбор и tail_from_end=false подберет пропущенные, но и продублирует все ранее собранные логи, поскольку новый файл -- новый поток (как минимум метка "имя файла" изменится).

Простое и однозначное решение не было найдено, но вероятность промаха мала -- оставлено нерешенным.

Слой api

Структура

В общем случае структура слоя api может быть описана следующей схемой:

• kernel содержит код, описывающий или дополняющий используемый протокол (основной код протокола чаще поставляется в готовых пакетах -- например, net/http). Конкретный протокол может содержать (или нет) сервер, абстрактный обработчик и DTO, роутер и другие элементы. Например, в http это http.Server, http.Handler, http.Request, http.ServeMux, а grpc предоставляет только grpc.Server, который совмещает несколько ролей, в т.ч. роль роутера.

• middlewares накладываются до передачи управления конкретному обработчику -- т.е. оборачивают все регистрируемые обработчики.

• handlers описывает конкретные обработчики.

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

  В некоторых протоколах обработчики могут быть определены с помощью описания контрактов (protobuf, openapi, ...). Реализация серверного контракта здесь называется service, чтобы не конфликтовать с термином "сервер", и выполняет роль внутреннего контроллера. В таких случаях сам обработчик может становиться неявным, даже если в kernel имеется соответствующая абстракция, поскольку контроль выполнения будет перехвачен сгенерированным кодом.

  Обработчики регистрируются и декорируются с помощью функции Register в слое сборки приложения.

• components содержит компоненты, используемые конкретным протоколом. Например, security в http или DLQ в kafcon.

На данной схеме цвет связей определяет устанавливающий их слой.

.proto-файлы

Генерация кода из proto-файлов настроена на пакет с именем pb -- для api/grpc/handlers/{handler_name}/pb. Поэтому в infra/clients/{service} пакет pb используется для удобства обновления клиента простым копированием файлов. Выделять подобный код в общие пакеты в данном проекте излишне.

Сервис может содержать несколько grpc-серверов и grpc-клиентов других сервисов. Имена proto-файлов, из которых был сгенерирован grpc-код серверов и перенесен grpc-код клиентов должны отличаться в пределах сервиса, иначе возникнет ошибка panic: proto: file ".proto" is already registered. Это значит, что, например, нельзя использовать файл с именем ".proto" для генерации сервера в сервисе auth и использовать для реализации клиента скопированные файлы из сервиса cfm, которые также были сгенерированы по файлу с именем ".proto". Соответственно, имена proto-файлов должны быть уникальны в рамках проекта.

OpenApi

Используется по аналогии с Protobuf для стандартизации и удобства использования структур запросов/ответов.

Каждый http-обработчик, если в нем используется openapi, содержит свою спецификацию -- это накладывает некоторые ограничения:

• необходимо дублировать общие компоненты, которые требуются спецификациям нескольких обработчиков:
  • папку common с общими элементами
  • components.securitySchemes из-за ограничений протокола и генератора (allOf и $ref).

В спецификациях сокращено количество описываемых http‑статусов до трёх базовых сценариев:

• 200 -- успешный ответ;
• 422 -- ожидаемые неуспешные сценарии (бизнес‑ошибки) с уточняющим полем code и текстовым сообщением message.
• редиректы (например, 307 Temporary Redirect).

Все прочие HTTP-коды в спецификации не описываются: они считаются внештатными ситуациями, багами и т.д.

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

KafApi

Kafka передает сообщения в бинарном виде, что делает необходимым наличие пакетов для десериализации сообщений в высокоуровневые структуры и определения контрактов их обработки по аналогии с Protobuf / OpenApi. Готовых инструментов описания контрактов и генерации кода найдено не было. Пример такого пакета -- admin_auth_events/kafapi.

Node Metrics Exporter

См. node/build/prometheus-exporter/README.md.

Kafka

Ограничение на добавление партиций

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

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

Консьюмеры

В идеальной среде в каждом сервисе-приемнике использовалось бы несколько консьюмер-контейнеров на каждый топик. В данном проекте для экономии ресурсов каждый сервис-приемник использует один консьюмер-контейнер на все топики, однако, разделение по топикам для наглядности выполнено с помощью отдельных горутин-консьюмеров (например, gateway). Стоит упомянуть, что kafka.Consumer может обрабатывать несколько топиков одним консьюмером, но это менее наглядно.

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

Неудачные идеи:

• Реализация многопоточного консьюмера, в котором основной поток-читатель передавал бы сообщения потокам-обработчикам, автоматически запускавшимся при ре-балансировке по одному на партицию, в конечном итоге была отклонена из-за несоразмерности выигрыша прилагаемым усилиям. Возможный выигрыш -- кроме примера fan-out-подобного кода для данного проекта -- гарантированная обработка каждой партиции в отдельном потоке, независимо от количества запущенных консьюмеров и партиций. Это ускоряло бы обработку сообщений в случаях неравномерного (не 1 к 1) распределения партиций по консьюмерам. Однако, потенциальное ускорение возможно лишь при равномерном заполнении обрабатываемых партиций -- в противном случае из-за заполнения буферов потоков-обработчиков (например, при достаточно медленной обработке и/или перекосе в партициях) такой консьюмер деградировал бы до стандартного однопоточного, блокируясь в ожидании передачи сообщений обработчикам. Также используемый пакет confluent-kafka-go основан на C-коде и не обладает детальной документацией / тестами (например, нет явной информации о том, какой метод в каких случаях возвращает какие ошибки), что осложняет поиск и применение решений, связанных с деталями реализации librdkafka (для той же буферизации потоков-обработчиков требуется информация о внутренних буферах, событии ре-балансировки и т.д.).

• Реализация параллельной обработки сообщений по группам из одной партиции еще более сложная (например, какой оффсет нужно сохранить, если группа Б обработана раньше группы А, а обработка группы А провалилась?), а проблема распараллеливания решается проще за счет увеличения кол-ва партиций и, соответственно, консьюмеров.

Гарантии доставки

Семантика доставки зависит от конкретного сервиса-потребителя / обработчика и топика, и может быть реализована как на уровне консьюмера с помощью сохранение оффсета (например, gateway), так и на уровне конкретного обработчика при необходимости смешивания различных семантик в рамках одного консьюмера (например, --- TODO ---).

ВАЖНО: В семантике at-least-once при передаче партиции в следствии ре-балансировки другому консьюмеру может возникнуть ситуация одновременной обработки сообщения несколькими консьюмерами (двумя или даже более при повторной ре-балансировке и, например, слишком медленной обработке сообщения), поскольку новый консьюмер-владелец партиции не ждет подтверждения ее передачи от прежнего владельца и начнет обработку с последнего сохраненного оффсета в группе -- т.е. как минимум с текущего сообщения.

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

В данном проекте из-за ограничения на добавление партиций и использования одного консьюмер-контейнера на группу (см. выше) ре-балансировка в следствии изменения состава консьюмер-группы исключена по умолчанию. Однако, ре-балансировка может произойти также и при достаточно медленной обработке сообщений в следствии превышения интервала max.poll.interval.ms, что делает проблему актуальной даже без необходимости в обобщении.

В общем случае можно попытаться решить проблему на уровне консьюмера изменением параметра max.poll.interval.ms и max.poll.records, но такой подход приемлем только при хорошо предсказуемой скорости обработки сообщений. Стоит учесть, что max.poll.records не поддерживается в librdkafka для Go, а kafka.Poll() поставляет сообщения по одному -- соответственно, max.poll.interval.ms определяет время обработки одного сообщения.

Можно попытаться ограничить работу обработчика с помощью отменяемого контекста, указав timeout/deadline в соответствии с max.poll.interval.ms и запасом времени на выполнение отмены операции, иначе отмена операции может выполняться уже одновременно с "прямой" обработкой сообщения другим консьюмером.

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

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

DLQ (Dead Letter Queue)

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

Представленный механизм DLQ основывается на базовом принципе работы Kafka -- группах сообщений: при попадании одного сообщения в DLQ все последующие сообщения той же группы автоматически направляются в DLQ -- это позволяет сохранить порядок сообщений в группе после устранения причины попадания группы в DLQ. Если же сохранение порядка не требуется, то идентификаторы групп уже должны быть уникальными для каждого сообщения.

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

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

Если скорость обработки сообщений из DLQ меньше, чем скорость их поступления в DLQ, следует остановить консьюмер до обработки всех сообщений необходимой группы (или всего топика). Программная блокировка в консьюмер-группе на время обработки сообщений из DLQ неприменима, поскольку вероятнее всего приведет к истечению max.poll.interval.ms и, соответственно, повторяющейся ре-балансировке. Вариант с установкой консьюмеров на паузу возможен, но неоправданно сложен по сравнению с ручной остановкой.

ВАЖНО: Остановка консьюмер-группы обязательна перед запуском DLQ-обработки при использовании несогласованной реализации хранилища сообщений и маркеров отравленных групп (см. dlq/storage.go).

Наиболее простым и универсальным способом возвращения группы сообщений из DLQ в нормальное русло является переотправка исправленных сообщений этой группы:

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

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

Повторная обработка сообщений

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

Debezium

См. init_admin_auth_events_producer.py

---