Шардирование сервиса объявлений Авито Доставки. Часть II

Привет, меня зовут Артем, и я работаю в Авито с 2016 года. Начинал как тестировщик, затем вырос в backend-инженера, с 2019 года пишу на golang, а сейчас руковожу командой разработки в Авито Доставке в роли техлида. Это вторая часть истории про шардирование сервиса объявлений Авито Доставки, где мы расскажем о реализации шардирования и полученном результате.

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

Первую часть статьи можно прочитать здесь.

Содержание:

• Реализация: как мы это сделали — на дворе 2023Q2

• 1. Создание шардированного хранилища

• 2. Реализация запросов с учетом шардирования

• 3. Тестирование шардированного хранилища 

• 4. Переход от одиночной к шардированной инсталляции 

• 5. Завершающий этап перехода

• 6. Какие результаты получились спустя 2 года эксплуатации решения

• Заключение

Реализация: как мы это сделали — на дворе 2023Q2

1. Создание шардированного хранилища

Само создание хранилища — дело довольно стандартное в компании, но подключение новой базы к старому сервису оказалось не так просто. У нас в инфраструктуре зашит принцип один сервис — одна база, и это хороший подход в целом, так как обеспечивает отсутствие «коммунальных баз», которыми пользуются несколько сервисов, что весьма полезно. Но есть нюансы: по такому принципу подключение второй базы к одному сервису также проблематично. Процесс занял несколько итераций общения с DBA, которые не давали большого результата.

В итоге помог бывший DBA, перешедший в бэкенд. Ниже — то, как он описал порядок необходимых действий:

Проделал как раз вчера с сервисом mod-duplicates-calculator, ему надо было доступ в базу сервиса mod-duplicates-search-ng (хранилище mod-duplicates-search-ng2).

Я добавил доступ:

access:
 services:
 - name: service-mod-duplicates-search-ng
 - name: service-mod-duplicates-calculator

Добавил пул:

pgbouncer:
  pools:
    - name: master
      dbname: master
      pool_size: 75
      pool_mode: transaction
    - name: mod_duplicates_calculator
      dbname: master
      pool_size: 10
      pool_mode: transaction

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

2. Реализация запросов с учетом шардирования

Дальше реализация запросов с учётом шардирования. Сначала подключаемся к кластеру, кластер — это по факту просто слайс подключений к базам, собственно нашим шардам.

type Cluster struct { _shards []*DB
    metrics *metrics.Metrics
}

Далее определяем интерфейс кластера, он состоит из 6-и методов.

type clusterInterface interface { 
    SharedByItemID(itemID int64) (*database.DB, error) 
    QueryxContext( 
      ctx context.Context,
      sharedMapping map[int64][]int64,
      buildQueryFunc func([]int64) (string, []interface{}),
      scanFunc func(rows *sqlx.Rows) error,
    ) []error
    ShardsMappingByItemIds(itemIDs []int64) map[int64][]int64 
    MainShard() *database.DB 
    Shards() []*database.DB 
    Close() []error 

1. ShardByItemId — по item_id (ключ шардирования) получаем подключение к шарду, где этот айтем лежит.

2. QueryContext — это самый интересный метод интерфейса, он гибкий и позволяет выполнять несколько запросов по ключам, мы используем его и для запросов по item_id и для запросов по user_id, для его использования нужно заполнить довольно сложные входящие параметры:

   • shardsMapping — мапа где ключ это номер шарда, а значение это слайс ключей, которые в нём лежат;

   • buildQueryFunc — функция, принимающая слайс ключей для запроса и возвращающая строку запроса и слайс аргументов для него;

   • scanFunc — функция, сканирующая и сохраняющая результат выполнения запроса, также возвращается слайс ошибок от шардов, если они есть.

3. ShardMappingByItemId — маппит айтемы на шарды, используется для создания мапы — shardsMapping.

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

5. Shards — отдаёт подключение сразу ко всем шардам и используется, когда мы не знаем, на каком точно шарде данные, у нас это запросы по user_id. Важно, чтобы их не было много!

6. Close — закрывает коннект к базе.

Далее приведу пример реализации запроса всех айтемов пользователя по user_id:

У нас есть текущая функция getAllByUserID. Для того чтобы научить её работать с шардированным хранилищем, нам нужно внутри неё определить функцию работы с шардами getAllByUserIDCluster:

1. shardsMapping — в данном случае мы каждому шарду присваиваем один и тот же user_id, так как не знаем, в каких шардах лежат айтемы пользователя.

2. buildQueryFunc — аргумент в запросе только один — это user_id.

3. scanFunc — каждый айтем добавляем в общий слайс — items, обязательно используем механизм синхронизации, так как запрос к каждому шарду выполняется в отдельной горутине.

Супер, мы определили основные параметры и готовы выполнить queryContext, посмотрим, что внутри:

func (cl *Cluster) QueryxContext(
    ctx context.Context,
    shardsMapping map[int64][]int64,
    buildQueryFunc func([]int64) (string, []interface{}),
    scanFunc func(rows *sqlx.Rows) error,
    ) []error {

for shardId, ids := range shardsMapping {
    wg.Add(1)

    go func(shardId int64, ids []int64) {
    defer wg.Done()

    db := cl.shards[shardId]

query, args := buildQueryFunc(ids)
rows, err := db.QueryxContext(ctx, query, args...)

for rows.Next() {
if err := scanFunc(rows); err

В коде, если упростить queryContext : сначала строится маппинг по shardsMapping, потом для каждого шарда запускается горутина с запросом, строящимся по buildQueryFunc, после чего собираются результаты по scanFunc. Получается довольно просто и гибко, один и тот же подход работает для запросов и по item_id и по user_id.

Таким образом мы подготавливаем все наши запросы к работе с шардированным хранилищем.

3. Тестирование шардированного хранилища 

Дальше идет этап тестирования шардированного хранилища. Рассмотрим, как все происходило — на дворе Q3 2023 года. 

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

Далее мы сделали мигратор данных, ниже приведу алгоритм его работы, базово он звучит так: возьми все данные из нешардированной базы и положи в шардированную)

Как это работает:

Заполню табличку так:

delivery_item.public> insert into items_moving
    select item_id from item where true limit 50000000
[2023-06-02 10:11:49] 50,000,000 rows affected in 24 s 442 ms

ПЕРЕД ДЕПЛОЕМ:

Складываем ИД всех айтемов которое есть в таблице item в табличку items_moving табличку хочу создать руками и после миграции удалить. Создание таблички:

create unlogged table if not exists items_moving (
    id bigint not null
);

ПОСЛЕ ДЕПЛОЯ РАБОТАЕТ ТАК:

1. Достаем из items_moving ИДишники айтемов, которые будем перекладывать в шардированную базу.

2. Достаем детальную информацию по каждому из этих айтемов.

3. Раскладываем айтемы по шардам.

4. В горугинах запускаем инсерт в каждый шард (если запись по айтему в шарде уже есть, то она попала туда «естественным путем», с ней не делаем).

5. Удаляем из items_moving то, что удалось заинвергнуть в шардированную базу.

Таким образом мы налили данных в шардированное хранилище, так выглядит работа мигратора на графиках.

Далее запустили в начале небольшую часть читающего трафика, примерно 30%, затем повысили до 60%, тут мы обнаружили, что нашей текущей коробки на 2 CPU уже не хватает, и повысили её до 4 CPU на шард и, наконец-то, дали все 100%.

• Первый тест — 12 июня случился первый тест на 32 шарда

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

Видим, что при 100% трафика транзакции выросли в 10 раз, CPU — в 8.

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

Нас это не устроило, так как в таком сетапе мы потребляли значительно больше количество ресурсов, чем хотелось бы. Мы экстраполировали полученные результаты работ нешардированной базы и базы с 32 шардами и получили прогноз роста TPS и потребления CPU. 

Исходя из расчёта видно, что уменьшив на конфигурации в 16 шардов мы получим рост общего CPU в 600%, то есть уменьшение общего CPU по сравнению с конфигурацией в 32 шарда почти в 2 раза, а нагрузка на один шард при этом вырастет незначительно, поэтому мы решили так и сделать.

При изменении конфигурации с 32 на 16 шардов мы словили первый LSR (критичный инцидент на продакшене). Дело в том, что у нас был кастомный healthcheck, который проверял доступность базы в каждом поде сервиса, и когда мы изменили конфиг базы, то половина шардов стала недоступна, а в сервисе конфиг остался прежним на 32 шарда и сервис ожидал доступности каждого из 32 шардов. Курьёзность ситуации ещё усугублялась и тем, что эта база никак не участвовала в пользовательских сценариях и её полная недоступность никак не должна была сказаться на пользователях, однако из-за этой проверки по readiness-probe сервис стал отвечать «503 Service Unavailable» на каждый запрос, что означало полный downtime.

Это также осложняло поиск корня проблемы, так как казалось, что проблема в k8s, однако мы оперативно продебажили код, нашли проблему и устранили, изменив также конфиг в сервисе, после чего всё пришло в норму. Отсюда можно сделать важный вывод — всегда надо критически оценивать, какой фукнционал точно должен работать, а каким можно пренебречь. Мы после этого, естественно, удалили данный healthcheck от греха подальше.

Итак, мы подключили конфиг из 16 шардов, очистили в них данные и налили снова. После этого мы подали на него нагрузку.

• Второй тест — 12 июля (спустя месяц после первого) мы запустили второй тест на 16 шардов

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

Снова сменили конфиг (уже без LSR) с 16 на 8 шардов, провели манипуляции по очистке и заливки данных и подали нагрузку, а тем временем на дворе наступил Q3 2023 года.

• Третий тест — 27 июля (спустя 2 недели после второго) мы запустили третий тест на 8 шардов

Результаты оказались сногсшибательными! В отличие от ожидаемого роста CPU на шард у нас случилось обратное, и утилизация CPU на шард снизилась по сравнению с конфигурацией в 16 шардов и стала равной конфигурации с 32 шардами, однако за счёт уменьшения кол-ва шардов в 4 раза мы добились также 4-х кратного уменьшения общего потребления CPU! 

Но всё же оставалось непонятно, почему так вышло, и мы пошли копать глубже.

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

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

Детально покопавшись, мы выяснили, что автоматика определила два шарда на одно и тоже ядро CPU и шарды конкурировали за ресурсы. Благодаря этому эксперименту коллеги из dba нашли баг в автоматике и пошли чинить.

Отсюда можно сделать ещё один важный вывод: при работе с БД важно понимать, что происходит на самих серверах, а не только в инстансе базы данных и на сервисе.

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

• повысить коробку до 8 CPU сразу, так как текущее потребление > 50% от 4 CPU;

• повысить память, сейчас она полностью утилизируется;

• оптимально распределить это по нашему железу.

Дальше уменьшать кол-во шардов мы не стали, так как с уменьшением кол-ва шардов у нас уменьшался и потенциал масштабирования нагрузки без необходимости решардинга, а текущая конфигурация нас устраивала по работоспособности и потреблению железа (не шардированная база потребляла 9 CPU, а шардированная стала потреблять 18.4 CPU).

Итоговая конфигурация — мы остановились на конфигурации в 8 шардов и 8 CPU на каждый шард (а планировали изначально 32 шарда по 2 CPU), что даёт 64 CPU лимит (как и планировали) при потреблении 18,4 CPU — другими словами мы утилизировали 28% из доступного CPU, что означало возможность увеличения нагрузки на CPU x2 без риска деградаций.

Настройки коннектов

Также во время экспериментов были обнаружены проблемы с коннектами, мы вывели детальные метрики по коннектам на стороне сервиса. В пакете database/sql есть функция, позволяющая вывести детальную статистику коннектов.

И обнаружили, что соединения долго висят в ожидании свободного коннекта, об этом свидетельствует не нулевой waitCount:

И также плачевную картину мы видели на стороне pgbouncer с постоянным фоном cl_waiting:

После нескольких экспериментов мы пришли к оптимальной конфигурации коннектов, полностью избавились от cl_waiting на pgbouncer и сильно уменьшили waitCount со стороны сервиса.

Распределение нагрузки по шардам получилось практически идеально равномерное:

4. Переход от одиночной к шардированной инсталляции  

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

Но тогда, как говорится, все карты уже были у нас на руках, данные писались в шардированное хранилище в асинхронном режиме и у нас был worker для переноса текущих данных в новое хранилище, то есть мы могли добиться практически 100% синхронизации данных на момент окончания работы воркера.

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

Для 1,6М RPM запросов различие было меньше 100 RPM или меньше 0,0000625% от всех запросов.

После этого мы подготовили детальный план перехода:

И план на случай, если что-то пойдёт не так:

DBA подтвердили план, также наш сервис отправлял данные в аналитическое хранилище и коллеги подтвердили, что с ним всё будет в порядке. Мы согласовали план со всеми заинтересованными лицами по проведению работ и стали ждать часа Х.

Итак, настал день Х, мы сделали, что могли, взяли в руки амулеты, помолились, попостились и приступили к переходу на шардированную инсталяцию.

• Первый переход — 12 сентября. Как говорится, первый блин комом, мы словили свой второй LSR: у нас появились долгие транзакции, из-за чего база не успевала обрабатывать все запросы также быстро, как раньше:

 И сервис начал деградировать:

Но мы воспользовались заранее подготовленным планом, быстро откатили изменения и всё восстановилось.

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

• Второй переход — 14 сентября. Как же мы ошибались, всё повторилось, как и в прошлый раз, но мы среагировали ещё быстрее и обошлись без LSR. 

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

Так как все эти данные уже были отправлены в аналитику из основного хранилища, то по факту их уже не стоит отправлять второй раз. Исходя из этого вывода мы придумали быстрый work-around, как это обойти, просто скипнув батч, который пытается отправится, тем самым остановив деструктивные действия для базы.

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

Также благодаря этому мы обратили внимание коллег из dba на особенности работы этого механизма, и ребята позже сделали его безопасным в подобных случаях.

• Третий переход — 18 сентября. Не зря цифра 3 считается «магическим» совершенным числом, в этот раз всё пошло так, как мы ожидали: снова начались длинные транзакции из-за отправки огромного батча в аналитику, мы применили наш work-around и скипнули батчи данных, после этого ошибки стали уходить, все запросы отрабатывали успешно, запросы с одиночной базы ушли полностью, сервис работал безупречно, а это значит, что мы, наконец, завершили переход на шардированное хранилище! И тем самым мы успели завершить цель в срок, до начала высокого сезона.

Подведем итог: мы заменили одиночную базу на 20 CPU на шардированную конфигурацию из 8 шардов каждый по 8 CPU, имели потребление CPU ~50% на одиночной базе в пике, получили от 28 до 35% потребление CPU на шардах. И в теории могли в дальнейшем применить горизонтальное масштабирование и увеличить коробку шарда с 8 CPU до 20 CPU, что давало нам большой запас в масштабировании нагрузки.

5. Завершающий этап перехода 

После успешного перехода работа над шардированием не заканчивается.

• Нагрузочное тестирование

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

Во время теста мы довели общий трафик до 3,6M RPM (1,6М органического трафика + 2М синтетической нагрузки, дошли до этапа, когда сервис начал деградировать, и остановили тест.

По метрикам базы видно, что мы снова уперлись в CPU:

В итоге мы подтвердили нашу гипотезу, что можем держать нагрузку х2 от текущей — а это целых 55k RPS без деградации! — и при этом можем расшить узкое место, легко повысив лимит CPU при необходимости через горизонтальное масштабирование.

• Обнаружили багу.

На дворе уже Q3 2023, октябрь, мы обнаружили проблему: при поиске объявлений с доставкой стали выдаваться товары выше 150к рублей, а такого быть не должно, так как у нас есть строгое ограничение, что доставку можно осуществлять только для товаров до 150к рублей. Например, айфоны с доставкой за 250к. Хотя доставки как таковой на товаре не было, это мешало поиску товаров с доставкой и вводило пользователей в заблуждение. Это наш третий LSR.

После расследования мы обнаружили проблему: при подготовке запросов в шардированное хранилище мы посадили баг: в запросе на удаление товара не возвращался item_id 

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

После осознания проблемы мы подготовили фикс, нашли все товары, которые потенциально могла затронуть данная проблема и актуализировали данные по ним. Также мы обнаружили, что тесты не покрывали данный сценарий, и исправили это, написав unit и integration тесты (e2e мы в команде не пишем, так как накладные расходы на их поддержку сильно высоки).

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

• Отказ от старой инсталляции

Тем временем на дворе был уже 2024Q1, январь: мы удалили старую базу и почистили свои хвосты, остававшиеся после перехода на шарды: почистили конфиги и освободили ресурсы базы с помощью dba.

Какие уроки мы получили

1. План может измениться самым непредсказуемым образом:

   • изначально мы хотели 2 шарда по 20 CPU;

   • dba хотели 32 шарда по 2 CPU;

   • в итоге сделали 8 шардов по 8 CPU.

2. Возникает много нюансов, которых изначально не ждёшь:

   • батчевые запросы на шарды приводят к сильному росту TPS и CPU;

   • необходимо детально настраивать коннекты на приложении и в pgbouncer;

   • могут возникать проблемы на самих серверах, на которых живёт база;

   • старый функционал стреляет в ногу (кастомные healcheck).

3. Шардирование помогает понять, где проблема, если сервис работает с шардами одинаково. А если шарды ведут себя по-разному, значит, проблема где-то на стороне шарда.

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

— Бу, испугался? Не бойся. DBA твой друг!

6. Какие результаты получились спустя 2 года эксплуатации решения

На дворе Q3 2025 года. Вот наши цели и факты:

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

Заключение

Вот что у нас получилось:

Плюсы:

1. Сервис работал исправно при текущей нагрузке, но был близок к пределу в 32k RPS и не имел возможности быстрого масштабирования, мы расширили этот предел до 55k RPS и имеем возможность как горизонтального, так и вертикального масштабирования при необходимости. 

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

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

Минусы:

Шардирование — это долго. У нас ушло 5 кварталов:

1. Q1 2023 — ресерч и выбор решения.

2. Q2 — создание хранилища, реализация обвязки.

3. Q3 — тестирование, переход.

4. Q4 — нагрузочное, багфикс, удаление старой базы.

5. Q1 2024 — чистка хвостов, закрытие техдолга.

Дополнительная сложность — больше кода, сложнее поддерживать и отлаживать сервис, так как реализация шардирования полностью на нашей стороне.

Благодарности

Помимо меня над задачей также работали коллеги из Доставки: Алексей Власов, Полина Харина, Роман Хурчаков.

И огромное спасибо всем кто помогал на этом пути, особенно:

• коллегам из DBA — за помощь и поддержку в переходе на шарды;

• коллегам из TechPR — за помощь в подготовке статьи.

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