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

В этой статье я расскажу, как устроен централизованный, надёжный и удобный способ асинхронного взаимодействия сервисов внутри Wildberries.

Я Виктор Такташов, Golang-разработчик в команде Trust & Safety. С задачами Highload знаком не понаслышке, потому что когда-то занимался сервером популярной онлайн-игры. К тому же я больше шести лет занимался борьбой, так что с проблемами расправляюсь умело и хладнокровно :)

Почему мы решили сделать Databus

Расскажу вам притчу о маленьком гофере, который пришёл в большую компанию. Я его даже для вас визуализировал:

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

Первая задача — понять, какие данные и контракты (соглашения о формате данных) вообще «гуляют» в компании. Гофер начинает обходить отдел за отделом. В компаниях вроде Wildberries много-много департаментов, поэтому гоферу приходится заглянуть в десятки разных дверей. Каждый департамент использует собственный кластер Kafka с особыми настройками.

Гофер тратит немало времени, чтобы узнать, у кого какие контракты, с кем интегрироваться, какие сообщения использовать. Но на этом задачи не заканчиваются.

Следующий шаг — найти владельцев всех этих Kafka-кластеров и договориться о доступах. У каждого подразделения свои правила: где-то доступ дают по запросу на почту, где-то через Telegram, где-то через корпоративный мессенджер… Гофер справляется и с этим. Спустя месяц он согласовывает доступы, собирает всё в кучу, пишет наконец свой сервис и запускает его.

Можно выдохнуть, правда?

Не проходит и пары недель, как к гоферу приходят со словами: «Всё сломалось, нужно чинить!» Он в недоумении, ведь всё работало как часы. Гофер снова идёт по отделам, общается с командами и выясняет, что одна из команд внезапно решила изменить контракт и поле UserId у них теперь будет не Integer, а String.

Гофер оперативно правит код, повторно выкатывает сервис — и всё снова работает. Но кто может гарантировать, что такая же проблема не будет возникать снова и снова?

Никто не может. Поэтому в Wildberries и назрела необходимость как-то централизовать зоопарк из множества Kafka у разных департаментов.

Основные требования к Databus

Первым делом мы сформулировали несколько ключевых принципов и требований к будущей платформе:

• Отказоустойчивость. Сервис должен быть доступен всегда и везде, даже под высокой нагрузкой. Никакой единой точки отказа — система должна продолжать работать при сбоях отдельных компонентов.

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

• Инфраструктура as a service. Пользователь не должен задумываться о внутренних инфраструктурных деталях. У него есть лишь два эндпоинта: один, куда отправлять события, и один, откуда их получать. Всё остальное решает платформа.

• Неизменяемость контрактов. Мы гарантируем стабильность форматов данных (контрактов). Если сегодня поле UserID имеет тип String, оно должно остаться String и через два-три месяца — никаких неожиданных изменений, ломающих интеграции.

• Коробочность и быстрый старт. Чтобы начать пользоваться Databus, разработчику не нужно писать кучу шаблонного кода или заполнять множество заявок на доступ. Достаточно подать одну заявку на подключение и импортировать нашу SDK-библиотеку — дальше можно сразу работать.

Сформулировав эти требования, мы перешли к реализации архитектуры.

Архитектура Databus

Сейчас всё устроено вот так:

События от отправителя (продюсера) поступают через входной прокси-компонент и попадают в кластер Redpanda (аналог Kafka). Консьюмеры получают события из кластера через выходной прокси. Между бизнес-сервисами и шиной данных стоят прокси-компоненты, которые мы сознательно внедрили вместо прямого доступа, чтобы скрыть весь внутряк от пользователя и эластично масштабировать входной и выходной трафик. При росте нагрузки мы просто поднимаем дополнительные экземпляры прокси.

Schema Registry необходим для валидации данных и управления контрактами. Он гарантирует неизменяемость схем: например, поле UserID должно быть строкой, а значит, никто уже не сможет отправить в него целое число. Schema Registry хранит все версии схем событий и позволяет обновлять их или добавлять новые.

Config Service отвечает за конфигурации — максимальный размер сообщения, адреса брокеров и пр. По сути, он представляет собой прокси для различных хранилищ конфигурации. У пользователя есть единый эндпоинт конфиг-сервиса, через который он получает все необходимые настройки. За самим конфиг-сервисом можно скрыть любую реализацию: сейчас в качестве хранилища конфигураций мы используем кластер Consul, но ничто не мешает заменить его хоть на обычные файлы на диске — для клиента ничего не изменится.

Databus Control Panel (DCP) — централизованная панель управления Databus, «точка правды» для всех схем событий и настроек системы. Подробнее будет в следующем разделе.

Поскольку единым хранилищем схем событий мы выбрали DCP, нужно было обеспечить рассылку обновлений схем всем участникам системы. ProtoSync — наша собственная утилита, которая автоматически отслеживает изменения схем в DCP и синхронизирует их со всеми остальными компонентами.

Кластеры Redpanda — «артерии», по которым текут события. Мы решили использовать Redpanda вместо классической Kafka, поскольку Redpanda предоставляет тот же API, что и Kafka, но при этом гораздо проще и удобнее в установке и эксплуатации.

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

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

Схема управляющей инфраструктуры выглядит так:

Панель управления DCP

Как я уже сказал, для управления всей этой инфраструктурой служит DCP — удобная веб-панель для пользователей и администраторов.

Возможности DCP для пользователей:

1. Разработчик может просматривать, как выглядит схема его события, и при необходимости загружать/обновлять схему через UI.

2. DCP интегрирована с системой мониторинга Grafana: прямо из интерфейса можно перейти к дашборду с метриками по своим событиям (в панели есть удобная кнопка для этого).

   

Возможности DCP для администраторов:

1. Панель разделена по окружениям (Dev / Stage / Prod), позволяя управлять схемами и сервисами отдельно на каждой среде.

2. Админы могут менять настройки Databus, наблюдать за состоянием кластеров и сервисов, просматривать системные метрики.

3. DCP предоставляет единый пульт управления всей шиной данных. Кроме веб-интерфейса, есть и API для автоматизации администрирования.

   

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

Для интеграции с другими частями инфраструктуры у DCP есть и программный API.

Консольная утилита WB-CLI

Помимо веб-интерфейса у нас есть консольный инструмент для работы с Databus — WB-CLI.

Некоторым пользователям удобнее выполнять действия через CLI, им больше по душе терминал. Поэтому мы параллельно с DCP развиваем и этот способ взаимодействия.

WB-CLI предоставляет ряд команд: авторизация, обновление и т. д. С помощью этих команд пользователь может в тестовом режиме подписаться на определённые сообщения (прослушать топик), отправить тестовое сообщение, увидеть список доступных событий. Также консоль позволяет управлять своими событиями: разработчик может самостоятельно создать новое событие в Dev-окружении и протестировать его, а затем самостоятельно же раскатать его на Stage и Prod.

Мы постоянно расширяем возможности WB-CLI, добавляя новые команды и функции — так же, как для веб-панели.

SDK Databus

Самая важная часть нашего решения — это SDK, библиотека для разработчиков, которая упрощает интеграцию с Databus. Сейчас SDK доступна для Go, Python и Java.

Эта библиотека собрала в себе множество типовых решений и примеров. Раньше каждая команда писала собственный шаблонный код для отправки и получения сообщений. Теперь же достаточно подключить готовую библиотеку: импортировать SDK и вызвать, условно, consumer.New().

SDK включает в себя множество полезных возможностей. Я перечислю лишь основные.

Механизм повторных попыток (Retry-транспорт). Сеть — штука нестабильная, в любой момент может произойти сбой. Наш SDK автоматически повторяет отправку/получение сообщений при сетевых ошибках. Реализация построена в стиле backoff: между повторными попытками выжидается пауза, увеличивающаяся со временем.

При этом предусмотрены проверки — например, если сервис вообще выключается (отменяется контекст), то повторять попытки не нужно. Также мы различаем типы ошибок: SDK не будет бесконечно ретраить при фатальных ошибках сервера, но попробует повторно при временных проблемах.

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

Автоматический offset management. Разработчику не нужно писать шаблонный код для того чтобы отмечать, какое сообщение обработано, а какое нет, — библиотека берёт эту рутину на себя. В данном примере можно увидеть, что после обработки сообщения SDK помечает его как прочитанное (даже если при обработке произошла ошибка декодирования, это учитывается).

TLS из коробки. До нашей реформы, чтобы подключить шину данных по защищённому протоколу, пользователям приходилось вручную загружать сертификаты, настраивать шифрование. Теперь достаточно добавить при инициализации опцию consumer.WithTLS() — и соединение с Databus сразу будет установлено по TLS. SDK под капотом автоматически проверяет сертификаты серверов.

Пакетная отправка и получение. SDK поддерживает работу с сообщениями через батчи. Можно собрать несколько событий в пачку и отправить их одним запросомодной командой. Аналогично можно получать сообщения блоками и на стороне потребителя.

Встроенные метрики. Сбор сразу встроен в библиотеку — разработчику не нужно думать, как отслеживать количество сообщений, ошибки, время обработки и пр. При интеграции SDK достаточно передать экземпляр структуры метрик, после чего библиотека начинает публиковать стандартные метрики системы.

Если нужно, пользователь может подключить и собственные обработчики или отправить метрики в дополнительное хранилище — SDK позволяет указать кастомные опции для этого.

Масштабирование: выбор схемы кластера

Одним из самых сложных вопросов при разработке Databus стала архитектура масштабирования и отказоустойчивости. Что использовать: один большой кластер или много небольших? Мы рассмотрели несколько вариантов, прежде чем прийти к финальному решению.

Первый вариант (самый сложный) предусматривал разделение на два кластера — пишущий и читающий — и использование специальных синхронизаторов между ними для репликации данных.

Мы довольно быстро отказались от этой идеи. Система излишне усложнялась, и к тому же появлялась точка отказа — те самые сервисы-синхронизаторы. Вдобавок дополнительные прослойки замедляли бы передачу сообщений.

Второй вариант — оставить разделение на два кластера (отдельно для записи и чтения), но добавить к ним резервные узлы. Однако и здесь проблема не исчезала: синхронизатор всё равно оставался единой точкой отказа., да и дублирование кластеров выглядело избыточным.

Третий вариант — отказаться от разделения ролей кластеров, но держать под рукой резервный кластер целиком. Рабочий кластер + запасной кластер, на который переключается трафик, если основной выйдет из строя. Такая схема повышает надёжность, но имеет свои минусы: фактически половина ресурсов простаивает в ожидании сбоя. Добавляет сложности синхронизация данных между кластерами.

Четвёртый вариант — самый простой и часто встречающийся: один кластер с набором резервных машин (узлов) внутри. При падении одного узла его роль берёт на себя запасной. Этот подход проще предыдущих, но нас смущало, что при катастрофическом сбое всего кластера система всё равно остановится.

В итоге мы остановились на схеме с множеством однотипных кластеров, каждый из которых имеет свои резервные узлы. Проще говоря, у нас развёрнуто несколько независимых кластеров Redpanda; данные распределяются между ними таким образом, чтобы нагрузка и риски были разделены. Если падает один кластер, это затрагивает лишь часть сообщений, трафик при необходимости перераспределяется на другие. При этом мы не вводим никаких дополнительных сложных компонентов — инфраструктура остаётся относительно простой и масштабируется горизонтально.

И ещё немного проблем

Масштабирование — только часть истории, всплыли и другие вызовы.

Довольно быстро стало понятно, что наши пользователи (внутренние команды) не слишком активно обновляют версии SDK. Если у них всё работает, переход на новую версию откладывается — вдруг при обновлении станет хуже. А что это значит?

Это значит, что в случае обнаружения багов исправления надо делать сразу в нескольких версиях библиотеки. Мы изначально вынуждены писать код, совместимый с более старыми версиями. Это, разумеется, замедляет внедрение новых функций и усложняет поддержку. Каждая новая фича должна либо не ломать старый интерфейс, либо сопровождаться выпуском обновлений для старых веток SDK.

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

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

Итого: тыква реально стала каретой

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

Теперь же появилось единое решение, которое берёт на себя всю эту инфраструктурную тяжесть. Разработчику достаточно подключить SDK и с её помощью отправлять или получать события — всё остальное платформа делает за него. Ему больше не нужно беспокоиться о надёжности доставки, форматах, хранении метрик и прочих подобных вещах.

P.S. Иногда нас спрашивают: «А у вас нет каких-то фич, как в обычной Kafka?» Databus изначально задумывался как несколько иное решение. Это платформа для передачи событий, которая закрывает примерно 90% стандартных задач и освобождает пользователя от всей инфраструктурной возни. Возможно, у нас нет некоторых редких возможностей, которые есть в «голой» Kafka, но если у команды вдруг возникает настолько хитрый кейс, что без специфической Kafka-фичи не обойтись, — что ж, вероятно, этой команде и правда лучше использовать Kafka напрямую. Наш Databus — это масштабное коробочное решение для большинства типовых сценариев, но не панацея на все случаи жизни. Пожалуй, в этом и есть его крутизна :-)