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

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

Содержание:

• Откуда растут ноги — на дворе 2023Q1

• Варианты масштабирования

• Сравнение вариантов масштабирования

• Шардирование. Что дальше?

• Промежуточный итог

Откуда растут ноги — на дворе 2023Q1

В начале 2023 года наш сервис объявлений Авито Доставки delivery-item обслуживал 1.6 млн RPM. Все данные лежали в одном Postgres-инстансе, который уже тогда хранил 148 миллионов айтемов. Нагрузка на базу доходила до 50% CPU из 20 доступных. При этом мы видели рост нагрузки и понимали, что база упирается в железо и нам необходимо придумать способ по масштабированию сервиса.

Какие цели мы себе поставили по масштабированию:

• держать x4 от текущего трафика (то есть до 6.4 млн RPM);

• рост объема данных x2;

• SLI > 99.9 при latency < 80 ms.

• а также важно выполнить всё до начала высокого сезона — до начала осени, то есть за 3 квартала.

Какие у нас были варианты решения:

• увеличить ресурсы базы — вертикальное масштабирование;

• кэширование (Redis);

• что-то поделать с Postgres: шардирование, партицирование, глубокий рефакторинг; 

• приделать новую базу данных (Redis, MongoDB, CockroachDB и т.д.);

• отказаться от хранения своих данных и перенести их в другой сервис;

• отказаться от хранения данных и рассчитывать все на лету;

• читать данные с реплик.

Мы начали по-честному исследовать каждый из этих вариантов.

Варианты масштабирования

Вертикальное масштабирование (увеличение ресурсов) 

В текущей конфигурации было 20 CPU, очень большая коробка, сейчас максимальная коробка — 8 CPU, но на тот момент можно было ещё использовать коробку в 30 CPU. Средняя нагрузка на проде — 24k RPS.

Что мы можем улучшить: 

• оптимизировать хранение данных; 

• уменьшить пишущую нагрузку;

• увеличить коробку с 20 до 30 CPU.

Однако у вертикального масштабирования есть свои проблемы: 

• упираемся в лимиты базы по CPU: даже увеличив коробку до 30 CPU, мы всё равно не сможем держать нагрузку x4, так как 10 CPU * 4 = 40 CPU, для нормального функционирования базы нужно держать утилизацию CPU ниже 50% от лимита, то есть нам бы понадобился лимит в 80 CPU.

• упираемся в лимиты pgbouncer по CPU на стороне сервера. Когда стали копать глубже в вертикальное масштабирование, выяснили, что серверный pgbouncer тоже является слабым звеном: он одноядерный и уже имеет максимальную конфигурацию, а его утилизация по CPU на тот момент  уже составляла 60%.

• упираемся в лимиты сервера по CPU. Как мы выяснили, нам необходима конфигурация базы с лимитом 80 CPU, но доступные сервера на тот момент предоставляли максимум 70 CPU, а это значит, что сам сервер, на котором крутится база, был бы под критической нагрузкой и испытывал бы трудности.

Вывод: этот вариант нам не подходит, так как упираемся в лимиты по CPU. 

Redis as cache

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

Для проверки нашей гипотезы мы провели эксперимент: сделали прототип решения по кешированию в redis по алгоритму «Сквозное кэширование» и пустили часть реального трафика через кэш.

Есть отличная статья «[По полочкам] Кэширование», в ней можно почитать подробнее про различные алгоритмы кеширования.

Примерно 12% трафика (далеко не весь) шло в кэш и наполняло его данными. На графике видно, как распределяются читающая и пишущая нагрузки в кэш: 

Состояние redis после часа записи трафика показывает, что мы закешировали 20М ключей из 148М айтемов в базе, то есть около ~15% от всех наших айтемов.

Далее мы измерили HitRate на стороне приложения, так как текущий redis использовался и для других целей, то мы не могли полагаться на его базовые метрики, и получили примерно 19% HitRate.

Если учесть, что в кэш идёт только 12% от всех запросов, то реальный HitRate = HitRate х 12% = 19% х 12% = 2%, то есть примерно 2% общего трафика шло в Redis и получало данные из кеша и не ходило в базу. При этом сервис в целом чувствовал себя хорошо и аномалий на метриках не было.

Мы закешировали 15% айтемов, пустили 12% запросов через redis, получили HitRate 19% и тем самым снизили нагрузку на базу на 2%. Если мы закешируем 80% айтемов (х5) и пустим весь трафик в redis (x8), то получим снижение читающей нагрузки на базу равное 2% х 5 х 8 = 80%, примерно настолько мы снизим нагрузку на базу.

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

Шардирование, партицирование или более глубокий рефакторинг

Начнём с конца: глубокий рефакторинг теоретически снижал бы нагрузку на 25% при записи и на 15% при чтении, что недостаточно и снова как и при вертикальном масштабировании мы упираемся в то, что всё должно крутится на одно сервере. Это не наш путь.

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

Шардирование

Далее мы рассмотрели несколько вариантов шардирования — разделения одной базы на несколько поменьше. 

• Первый — это ручная реализация с использованием модуля postgres_fdw. 

postgres_fdw простыми словами — это секционированная таблица, где секции — это внешние таблицы, которые могут располагаться на других серверах, то есть выступать в роли «шарда». 

Из плюсов — это уже есть в ядре Postgres, и не нужны изменения в коде.

Но из минусов — все запросы происходят последовательно (postgres_fdw открывает только одно соединение для каждого «шарда» и выполняет все запросы к нему последовательно), что дает большой оверхед на сеть. Соответственно, этот вариант получается плохим с точки зрения latency. 

• Второй вариант — это полностью ручная реализация алгоритмов шардирования на стороне сервиса. 

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

Но из минусов — писать много кода. 

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

1. Возможность такого способа шардирования.

2. Нет ли очевидных деградаций.

3. Какие таблицы шардировать и как хранить остальные.

Чтобы найти ответы на эти вопросы, мы пошли делать прототип решения: 

1. В прототипе мы реализовали шардирование на основе первичного ключа item_id и линейного алгоритма шардирования, роутинга данных по шардам как остаток от деления item_id % кол-во шардов. После реализации мы раскатили решение на тестовый контур, добились успешного прохождения всех наших автоматических проверок unit и integration тестов, убедились в том, что все наши API отвечают и все асинхронные задачи выполняются успешно. Это позволило дать ответит «да» на первый вопрос. 

2. Далее мы запустили нагрузочные тесты на отдельном perf-контуре в 1k и 5k RPS (нагрузка пропорциональна конфигурации базы, то есть база меньше, чем на проде, и нагрузка меньше) и по их результатам никаких деградаций не выявили.

3. У нас есть основная таблица, наиболее жирная, где хранятся все важные данные по айтемам и несколько дополнительных-служебных таблиц, тут мы также рассмотрели 3 варианта, забегая вперёд мы выбрали первый вариант, прототип был также построен на нём, так как не используем джойны, что нивелирует основной его минус и остаются только плюсы: 

   • первый вариант — шардировать только основную таблицу, а остальные хранить на одном шарде;

   • второй вариант — шардировать все таблицы: плюсов нет, главный минус — усложнение кода;

   • третий вариант — шардировать только основную таблицу, а остальные хранить на каждом шарде.

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

Новая база данных (CockroachDB) 

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

Ребята взяли все возможные решения, выбрали 4 наиболее перспективных: MongoDB, CockroachDB, Cassandra/Scylla и Tarantool. Сравнили их в разрезе десятка наиболее важных параметров с PostgreSQL, в итоге остановились на CockroachDB, так как она поддерживает все необходимые продуктовые кейсы, полностью совместима с PostgreSQL и является cloud native, из минусов выделили только некритичное падение производительности. При детальном рассмотрении этого варианта мы выяснили, что полноценная поддержка новых БД данного типа ожидается только через три квартала, а мы не можем столько ждать, так как за это время нам уже необходимо реализовать наше решение, чтобы закрыть потребности бизнеса.

Вывод: данное решение нам не подходит.

Отказаться от хранения данных (перенос в другой сервис)

Мы рассматривали идею передать доставочные параметры в главный сервис объявлений Авито — service-item, они ведь и так хранят очень много данных, почему бы не хранить ещё и данные Авито Доставки? Такой тезис мы решили проверить: конечно, слепо верить в это было бы наивно, но и не проверить мы не могли, а потому предложили это команде service-item. Ожидаемо, команда item-ов отказала с очевидными аргументами: это превращало бы их сервис в монолит, который собирает в себя всё подряд.

Вывод: данное решение нам не подходит.

Отказаться от хранения данных (полностью)

Идея вообще не хранить данные самая радикальная и простая, нет данных — нет проблем, возникающих при их хранении. Логично же, верно?

Проблема в том, что для каждого запроса пришлось бы ходить по всем зависимостям, на основе которых наш сервис формирует данные, это десятки других сервисов и они не приспособлены под наши большие нагрузки 1,6М RPM и просто не выдержали бы их, а масштабировать десятки сервисов задача ещё сложнее, чем наша текущая. Данные также нужны для аналитики, а аналитическое хранилище само по себе ненадёжное.

Вывод: данное решение нам не подходит.

Читать данные с реплик

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

Вывод: данное решение нам не подходит.

Сравнение вариантов масштабирования

На выходе у нас получилась такая табличка, где наглядно показано сравнение всех вариантов решения:
 

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

В итоге поняли, что нам подходят два варианта — кеширование и шардирование. 

Выбрали шардирование:

• оно масштабирует и читающую, и пишущую нагрузку, а также размер хранимых данных;

• не добавляет новых зависимостей;

• не требует отказа от хранения;

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

Шардирование. Что дальше?

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

• выбор ключа шардирования;

• выбор алгоритма шардирования;

• выбор конфигурации шардов;

• план реализации;

• реализация.

Ключ шардирования

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

1. Равномерное распределение данных, чтобы не было «перекоса» в шардах.

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

3. Большинство запросов должно идти по ключу, чтобы избежать кросс-шардовых запросов.

У item_id больше уникальных значений, что даёт более равномерное распределение нагрузки по сравнению с user_id, так как обычно у одного пользователя много товаров, это также обеспечивает и более равномерное распределение нагрузки, к тому же у нас 99% запросов идут по itemId, по userId — менее 1%. Мы также уже реализовали прототип на item_id, поэтому выбор был прост — мы остановились на item_id.

Алгоритм шардирования

Критерии к алгоритму:

• равномерное распределение данных;

• скорость поиска;

• простота и надежность.

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

Из таблички выше очевидны два победителя: линейный алгоритм и maglev. Так как третий наш критерий — это простота и надёжность, то мы выбрали простой линейный в виде остатка от деления item_id на кол-во шардов. Да, он не поддерживает простой решардинг, но он у нас не в приоритете.

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

Конфигурации шардов

Тут мы изначально хотели сделать 2-4 шарда, так как считали это наиболее оптимальной конфигурацией по своим внутренним соображениям, а эксперты из DBA хотели наоборот, чтобы мы сделали «много маленьких шардов», так как чем меньше БД — тем проще её поддерживать

Итого: изначально выбрали 32 шарда по 2 CPU (с возможностью вертикального роста до 30 CPU).

План реализации

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

1. Создание шардированного хранилища. Для этого необходимо сделать заказ инсталляции и настройку коннекта к ней с помощью dba. Обязательно стоит договориться с dba о поддержке как можно раньше и заложить необходимые ресурсы. Для этого у нас в компании используется техника TDR’ов — Technical Design Review, где на этапе ресёрча/проектирования можно пригласить необходимых экспертов и согласовать все нюансы, что мы и сделали, поэтому тут риски нивелировали сразу.

2. Реализация запросов с учётом шардирования. Вся логика шардирования реализуется на стороне приложения: выбор шарда, подготовка запроса, выполнение запроса и сбор результата. Это всё мы делаем своими силами, поэтому считаем что рисков тут нет. Всё сами запилим, все компетенции есть в команде. Привет bus-factor и подобное, но тут мы это в расчёт не берём.

3. Тестирование шардированного хранилища. Для полноценного тестирования нужно сделать много приседаний, так как мы хотим проверить полное соответствие «боевому» положению. Для этого нам надо: написать мигратор данных, провести эксперименты по наполнению шардов данными и подать запланированную нагрузку. Компетенции все уже есть в команде и всё это в целом казалось делом нехитрым, но, как выяснили позже, данный этап занял по времени столько же, сколько и два предыдущих, даже с учётом того, что частично делался в параллель реализации запросов.

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

5. Отказ от старой инсталляции. Когда успешный переход свершится, важно «убрать за собой»: планируем следующие действия — очистка конфигов в сервисе, удаление лишнего кода, освобождение ресурсов с помощью dba.  

Итак, план готов, как говорят классики:

— У вас была какая-то тактика?

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

Промежуточный итог

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

Масштабирование одной Postgres-базы — 1.6 млн RPM, 148 млн записей и 50% утилизация CPU при необходимости держать нагрузку x4.

Перебрали всё: вертикальный скейл, кэширование, новые БД, чтение с реплик и даже отказ от хранения. Работоспособными оказались только Redis и шардирование, но кэш не решал проблему записи.

В итоге выбрали ручное шардирование на уровне приложения: item_id как ключ, линейный алгоритм (item_id % количество шардов), шардируем только основную таблицу. Это дало масштабирование, чтения и записи без новых зависимостей, а также получилось уложиться в сроки — при чётком плане миграции и тесной работе с DBA.

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

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