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

Про них нам расскажет Могилин Виктор, руководитель группы разработки стораджей в Почте Mail.ru (компания VK). Он опишет, что такое объектный сторадж в деталях, а также поделится своим опытом в реализации такого хранилища.

Почтовый сторадж

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

Проблема

Довольно давно сторадж был устроен таким образом, что почтовый ящик юзера был реализован просто как папка на диске, внутри которой лежали все письма и индексы. Горизонтально масштабироваться было легко, потому что можно было доставлять столько дисков сколько хочется. Но в определённый момент стало ясно - горизонтальное масштабирование имеет свои ограничения. Когда в почте было уже около 3 тысяч серверов с данными, стало ясно, что они банально занимают слишком много места в дата-центрах. И их обслуживание начинает представлять собой нетривиальную инженерную задачу. Встала проблема вертикального масштабирования - увеличить объем дисков, а также вешать больше дисков на один сервер, чтобы в итоге снизить количество серверов. Но чем больше диск, тем больше на нем данных, а, следовательно, и больше обращений (операций чтения и записи).

Современные HDD диски вытягивают примерно 200 IOPS (причем, это не зависит от объема). Но когда на них собирается много папок и писем (100 тыс. – 1 млн.), каждая дисковая операция становится дорогой. Дело в том, что сначала нужно прочитать метаданные директории, заполнить dentry кэш, загрузить inode в кэш файловой системы, что — тоже грязное чтение с диска, и только потом приступить к чтению писем. При этом записи файлов и апдейты индексов — это постоянные рандомные сики (seek, т.е. операция позиционирования головки) в диск, которые магнитные диски не любят.

Задача

Разработчикам Почты Mail.ru пришлось сокращать число серверов на порядок с 3-х тысяч до примерно 3-х сот. Но история с сокращением превратилась в целый сериал из трёх серий. Потому что к разным частям письма нужен разный паттерн доступа.

В первой серии команда Виктора оптимизировала хранение вложений (Attach). По IOPS вложения оптимизировать нет смысла, но можно заморочиться с их дедупликацией. Подробнее о том, что это такое, вы можете узнать из доклада Андрея Сумина.

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

При этом было два варианта внешнего диска: HDD или NVME.

NVME диски вытягивают тысячи IOPS, но много их в сервер не запихнуть. А команда Виктора хотела, чтобы в одном сервере было примерно 100 дисков по 18Tb, чтобы в сумме получилось больше 1Pb. Поэтому им пришлось выбрать HDD и решать проблему IOPS.

Они заметили, что структуру письма можно распарсить на входе, то есть структуру портов положить в индексы. А тело письма рассматривать как BLOB, бинарные данные без всякой структуры, которые можно один раз положить в сторадж и оттуда читать или удалять.

Принципы, по которым стоит хранить BLOB, известны. На этих принципах строятся все современные объектные стораджи. С этого и началась история собственного Mail.ru объектного хранилища Zepto.

Устройство Zepto

Zepto состоит из трёх компонентов:

zeptoproxy — точка входа в кластер;
statusservice — хранилище метаданных кластера;
bucket’ы — это то, что в аналогичных системах называется shard’ом, image или volume. В bucket’ах хранятся данные, а общение с ними происходит через bucketservice.

Подробнее о bucket

На физическом уровне bucket — это немодифицируемый файл, в который можно дописывать только в конец. Когда bucket создаётся, на него вызывается fallocate — syscall, который поддерживается в современных файловых системах и ядрах. Он гарантирует, что при успешном выполнении для файла выделяется пространство с максимально последовательными блоками на диске. Гарантируется, что во время записи в такой файл не случится ошибки из-за отсутствия места, так как система заранее зарезервировала место на диске под него.

Структура bucket’а следующая: в начале стоит служебный заголовок, потом идут пользовательские записи одна за другой, а в конце — специальная запись End Of File (EOF).

При этом пользовательские записи записываются страничками по 4Kb и у каждой из них есть небольшой заголовок и перечисление характеристик: размер, сжат, не сжат и recordID.

RecordID и версия

RecordID записи — это смещение от начала bucket’а до начала записи. Если recordID записи равен 100, значит, надо взять bucket, сделать seek на 100 байт, и тогда можно читать пользовательскую запись.

Здесь же вводится важное понятие — версия bucket’а — это смещение до End Of File. Оно нужно для того, чтобы закрывать bucket на запись, а не писать его до бесконечности. Например, если мы хотим держать bucket’ы до 2-х Gb и чуть больше, то просто закрываем их на запись, когда версия достигает нужного значения.

При этом 4Кb странички снабжаются crc кодом, который проверяется при каждом чтении. Кроме того, у Mail.ru есть механизм, который «ползает» по bucket‘у и сверяет crc суммы — таким образом, можно обнаруживать сбойные диски и восстанавливать данные.

bucketservice

Чтобы читать bucket’ы и в них писать, над ними поднимается bucketservice на диске, через который осуществляется всё общение.

Таким образом, сервер выглядит так: 1 диск = 1 bucketservice. То есть над каждым диском поднят bucketservice. При этом все bucketservice’ы имеют одинаковый IP, но разные порты. Это позволяет обратиться к любому bucket’у внутри этого сервиса.

bucketcompactor

Чтобы выполнять фоновые джобы, компакты, crc чеки и создавать bucket’ы, есть отдельный фоновый процесс bucketcompactor:

To` он выполняет служебные задачи. В Mail.ru его часто лимитируют через cgroups, чтобы он не кушал лишних ресурсов.

statusservice

Statusservice — это хранилище метаданных:

Она примерно раз в минуту опрашивает все bucketservice’ы. Во время первого запроса получает всю информацию о них, а потом — изменения в виде дельт.

В памяти statusservice держит карту bucket’ов: где они находятся, какая у них версия и, например, сломан он или нет.

Также statusservice предоставляет API для zeptoproxy. В частности, statusservice выполняет два основных запроса: ищет по ID адреса bucket’ов, из которых нужно выполнить чтение, а также выдает пригодные для записи bucket’ы.

Ещё statusserivice занимается мэйнтенингом: поддержкой в кластере определённго количества bucket’ов доступных для записи, планировкой сверок контрольных сумм и другими служебными задачами.

Схема добавления дисков в кластер

Необходимо решить задачу: откуда statusservice узнает, где находятся bucketservice, которые необходимо опрашивать. Zepto решает этот вопрос тем, что вводит определенную схему добавления дисков в кластер, которая жёстко и фиксировано задаётся при инициализации кластера. В ней описано, как должны быть соединены диски между собой при добавлении в кластер. Они добавляются группами. Вот пример, как это происходит для фактора репликации равного x2 в 3-х дата-центрах:

Минимальный набор дисков для группы — 6 штук. Они организовывают между собой 3 пары так, чтобы диски внутри каждой пары находились в разных DC.

Если третий дата-центр вылетает, то остаётся одна пара. В неё можно продолжать записывать (disk1-disk3), а данные, которые находились на дисках в DC3, можно читать с реплик. Они остаются живыми в 1-м и 2-м дата-центрах.

Для репликации x3 в 5 дата-центрах минимальная группа — это уже 15 дисков. Соответственно, также больше связей и выше надёжность.

В такой схеме можно потерять два DC и сохранить доступность на чтение. Даже если вылетит DC4, то по-прежнему останется тройка, в которую можно писать.

Организация диска

На самом деле bucketservice работает не над дисками, а над партициями. Если реплицировать диск в диск, то если вылетит диск, все read, которые могли бы туда прилететь, прилетят на реплику. Таким образом реплике может стать не очень хорошо. Но если диск разбить, например, на партиции по 2Tb и организовать из них группы, то при вылете диска можно размазать нагрузку.

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

zeptoctl

Для управления кластером есть служебная утилита zeptoctl. С ее помощью при инициализации кластера задаётся схема шардирования для statusservice (для отказоустойчивости в боевом кластере необходимо держать более одного statusservice). Так же, с ее помощью настраивается схема групп, которыми в кластер добавляются новые диски. Еще пример применимости zeptoctl - операция move in: когда добавляется новая группа из чистых дисков в кластер, то эти они имеют больше записей, чем другие. То есть в первую очередь записывают недозаполненные диски. Если в них делать много записей, то потом может получиться перекос по get’ам, потому что к новым письмам обращаются чаще, чем к старым. Соответственно, zeptoctl move in переносит часть данных кластера на новые диски, чтобы нагрузка более равномерно балансировалась бы на все диски. Сложных весов для балансировки Zepto не использует, вместо этого просто отслеживает средний disk_usage по кластеру и по средней заполненности выравнивает новые диски.

Также есть простенький язык запросов, который позволяет селектить bucket’ы. Селектить bucket’ы можно как по дата-центрам, так и с определённым статусом. Вот как это примерно выглядит:

Команды Zepto

У кластера по большому счёту три команды:

GET — достать запись;
PUT — положить запись;
DELETE — удалить запись.

Здесь уже возникает небольшое отличие от стандартных схем. В Zepto нет выделенного слоя API, роутера, поэтому zeptoproxy ставится прямо на сервер клиентам, которые сохраняют данные и получают их из хранилища. Таким образом можно гонять меньше данных по сети и применять ещё некоторые оптимизации.

PUT

Чтобы вставить в кластер какие-то данные, клиенту нужно передать их через zeptoproxy, который в рандомном порядке опрашивает statusservice и с помощью команды lock пытается получить bucket для записи. Ему отвечает statusservice: предоставляет информацию о bucket’е, его репликах и версии. В этот момент bucket lock’ится на запись внутри этого statusservice и пока lock не будет снят, другой клиент этот bucket не получит. Zeptoproxy параллельно записывает данные во все реплики, и, если всё успешно, снимает lock.

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

В результате записи клиенту выдается специальный ID, который складывается из bucketID и recordID. По этому ID в дальнейшем можно будет прочитать запись.

Получается, что в Zepto идентификатор объекта (ID) - это число (зачастую в других объектных хранилищах используется длинная строка), с которым удобно работать, потому что его можно компактно хранить в индексах (можно кодировать всякими Varint, ZigZag). Еще один плюс - не нужен дополнительный индексный файл, который должен был бы хранить соответствие идентификатора адресу и смещению записи.

Минус в том, что Zepto ID - это 64 битное число, т.е. количество записей в такой сторадж ограничено. Но, с другой стороны, 64 бита позволяют адресовать до эксабайта, это довольно много, учитывая то, что в почте на текущий момент всего 15PB писем (это без учета аттачей).

И всё же что-то может пойти не так практически на любой стрелке, но везде работают retry’и. Если не получается взять lock в одном statusservice, то система переключается на другой; если не получается записать кворум во все реплики, то система начинает всё сначала, т.е. попросит другой бакет в другом statusservice. Причём есть разная логика выбора bucket’ов из других дата-центров, чтобы не пытаться в один и тот же сохранять. Если не получается закоммитить запись (т.е. подтвердить в statusservice, что запись прошла успешно), то опять же процесс записи начинается сначала.

Если в кластере есть “здоровая” пара дисков, то процесс когда-нибудь процесс сойдётся.

GET

Схема выглядит довольно просто:

Клиент передаёт zeptoID в zeptoproxy, из него берется bucketID и делается запрос на любой statusservice в рандомном порядке. statusservice возвращает IP адреса, где лежат реплики bucket’а, и запись читается из одного рандомного bucketservice. Если что-то пойдёт не так, то какое-то количество байт можно прочитать из одной реплики, а остальные байты — из другой. Retry’и здесь работают так же, как в PUT.

DELETE

Обычно delete делается через внешние БД или через журнал удалений. В Zepto delete — это put специальной записи в bucket. Причём такие delete вставляются даже в уже закрытые на запись bucket’ы.

Удаление происходит не мгновенно. Если в bucket’е было 3 записи, и первые 2 из них удалились, то какое-то время bucket в таком виде будет существовать пока компакт полностью не перезапишет bucket. Таким образом данные удаляются физически.

Если удалить 2 записи из bucket’а, то все recordID, смещения внутри bucket’а не будут соответствовать реальному положению дел. Чтобы эту ситуацию обрабатывать, в Zepto есть индекс удалений в bucket’е. Он хитро устроен, чтобы поменьше весить. Грубо говоря, это специальная map, в которой удалённым recordID и записям соответствует их offset. Так Zepto корректирует доступ к bucket’у.

Чтобы удалить данные из диска, к которому по непонятным причинам нет доступа, нужно положить удаление в любой доступный writable bucket. Когда компакт доберётся до этого delete, он поймёт, что это delete от другого bucket’а и будет бесконечно пытаться доставить его до адресата. Такие delete’ы называются foreign delete, они нужны для отказоустойчивости удалений во время сбоев.

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

За счёт чего уменьшается IO?

Сами файлы довольно большие, и, допустим, в портиции 2 терабайта, таких файлов получится не очень много, примерно тысяча штук. Zeptoproxy все эти bucket’ы хранит в одной папке, то есть в одной папке хранится около тысячи файлов. При этом dentry кэш никогда не вымывается, так как inode (ввиду их малого количества) практически всегда закэшированы. Поэтому read выполняется за одну IO операцию, а write — фактически бесплатен, потому что всегда идёт в конец открытого заранее bucket’а, у которого аллоцированные блоки на диске. IO при записи, по сути, тратится только тогда, когда головке диска нужно перепрыгнуть с одной дорожки на другую.

За счёт чего достигается надёжность?

Надежность - это способность не терять данные в случае выхода диска из строя. В Zepto надёжность достигается за счёт репликации. Хотя bucket — это просто физический файл, но система работает с ним как с логической сущностью, то есть как с объектом, у которого есть bucketID и набор физических реплик (2-3 штуки).

Создание bucket’а инициирует statusservice. Он делает это, когда не хватает writeable bucket’ов.

Сначала statusservice генерирует новый ID, потом посылает две команды в два разных дата-центра. Там bucketservice с baketcompactor создают новый файлик. Если всё проходит успешно, statusservice записывает к себе в память, что появился новый bucket с ID, IP и версией. При этом можно начинать записывать файл, начиная с нулевого смещения, то есть 0-ой версией.

Когда zeptoproxy получает bucket для записи, то она также получает адреса всех реплик, в которые выполняет кворумную запись.

Zepto поддерживает следующие факторы репликации:

x2
x3
x1.5

С первыми двумя всё понятно: у каждого bucket’а 2 или 3 реплики и кворумная запись в 2 или 3 места. Фактор x1.5 — это “экономный” режим хранения, обычно в объектных стораджах его называют erasure coding.

bucket создаётся в 3-х экземплярах в 3-х разных дата-центрах, но третий bucket для начала является заглушкой, а запись идёт в 2 bucket’а, как в обычной схеме с x2-репликацией. Когда bucket закрывается, запускается специальный процесс, который в команде Виктора называют «полторирование», потому что он полторирует bucket: на одном диске оставляет чётные биты bucket’а, на другом нечётные, а на третьем — XOR чётных и нечётных. За счёт этого достигается x1.5 фактор репликации.

В таком кластере 100 байтный файл занимает 150 байт, по 50 на каждом диске. В схеме с x2-репликацией 100 байтный файл занимал бы 200 байт (по 100 байт на каждом диске).

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

В итоге, надежность в Zepto достигается за счёт:

Репликации;
Постраничного crc, который мы всегда проверяем;
Возможности обнаружить и выполнить fix «сломанных» бакетов.

При этом bucket’ы не очень большие (2-3Gb), поэтому “сломанный” bucket всегда можно довольно быстро переписать со “здоровой” реплики.

За счёт чего достигается доступность?

Под доступностью понимается возможность чтения/записи в систему при вылете дата-центра из строя. В Zepto есть 2 уровня, на которых нужно обеспечить доступность: statusservice и bucket’ы (слой хранения).

Без statusservice ничего работать не будет, потому что только он знает, где какие bucket’ы лежат и может выдать bucket для записи или чтения. Поэтому statusservice нужно как-то дублировать.

Существуют разные подходы в разных объектных хранилищах. Иногда ставят реляционную БД и организуют синхронную репликацию. Иногда ставят eventual consistency систему, наворачивают рафт и автовыбор мастера. Но и то, и другое решение подразумевает даунтайм (на запись) при вылете мастера. Так как команда Mail.ru этого хотела избежать, она сделала statusservice’ы независимыми друг от друга, то есть каждый из них опрашивает все bucket’ы и имеет свою карту bucket’ов в памяти.

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

Получается, что каждый такой statusservice является “мастером” для своего диапазона. При этом информацию для чтения о bucket’е можно получить из любого statusservice’а. А если какой-то statusservice упадёт, то в кластере просто станет меньше writeable bucket’ов, но запись будет по-прежнему доступна - ведь пока жив хотя бы один statusservice, он будет генерировать и выдавать на запись новые bucket’ы.

Доступность на слое хранения данных обеспечивается за счет физического разделения реплик bucket’ов на разных серверах в разных дата-центрах.

Максимальный уровень доступности - x3.

Резюме

Итак, к чему же в итоге привела работа над хранилищем.

На первом этапе почтовые вложения были вынесены со стораджей во внешнее хранилище, а также дедуплицированы.

На этих же серверах команда развернула Zepto для хранения самих писем.

Сетап получился следующим: есть сервер с 12-ю дисками, к нему же подключается SATA-полка, на которую подключается ещё 60 дисков (все диски 18TB). Zepto и вложения делят между собой место (примерно 6-8TB под Zepto, остальное - вложения). Диски хорошо утилизируются по месту. По IOPS получается приличный запас. То есть, можно будет использовать диски еще большего объёма, когда они появятся.

При этом, письма в Zepto хранятся с x2 фактором репликации и, казалось бы, 15PB должны превращаться в 30PB, но Zepto также использует сжатие, а письма очень хорошо жмутся, поэтому после сжатия 30PB обратно превращаются в 15PB.

В почте Mail.ru нет мегахранилища, в котором хранится всё. Вместо этого применяется подход, когда для определенной задачи поднимается свой “маленький”, независимый кластер Zepto. К примеру, есть такие кластеры:

Облако 80PB, фактор репликации x1.5;
Сниппеты для поиска по почте 2.4PB, фактор x2;

В кластере с письмами средняя скорость доступа сравнима с доступом в диск. В случае облачного кластера там, во-первых, большие объекты, а во-вторых, фактор x1.5, т.е. надо читать с двух дисков и “склеивать” файлы перед отдачей клиенту; как следствие, цифры поскромнее.

Фактически, скорость Zepto-кластера зависит от дисков, которые стоят под Zepto, и от фактора репликации.

Из нереализованного, клиенты часто просят гибридную схему репликации, когда кластер работает в режиме репликации x3, но при этом диски не переходят в режим read only с потерей одной из реплик, а “временно” начинают работать в режиме x2. При этом еще желательно уметь перемещать “холодные” данные в рамках того же кластера в режим x1.5.

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

Из недостатков, пожалуй, можно выделить особенность, как сделаны удаления. За счёт того, что они всё-таки прилетают не только во writeable bucket’ы, но и в уже закрытые на запись (удаление старается попасть в тот же bucket, где лежат и сами данные, которые надо удалить) - это порождает random IO в диск. С большим количество удалений это сделает кластеру больно. В планах все же реализовать и альтернативный механизм удалений, например, через журнал.

Еще в планах сделать больше автоматики. Сейчас, например, Zepto умеет многое аллертить (о сбоях диска, о побитых crc-суммах, о сбоях внутри кластера), на эти алерты обращают внимание сисадмины и обычно чинят вручную, запуская те или иные функции zeptoctl. Это оправдано боязнью админов, что автоматика сойдёт с ума и потрёт данные. Но что-то можно на самом деле автоматизировать, например, “развоз” всех целых bucket’ов с диска, если на этом диске много IO-ошибок, и “выкидывание” такого диска из кластера.

Заключение

Довольно много типов данных можно рассматривать как BLOB (почтовые письма, старые логи, поисковые сниппеты). Одна из лучших схем хранения для BLOB - это объектные хранилища, они способны обеспечить эффективное хранение, быть надежными и доступными. Выше рассмотрены основные принципы построения объектных хранилищ на примере Zepto - хранилища от Mail.ru.

Zepto позволила сократить число серверов.

Изначально было 3К серверов. В первую очередь дедуплицировали и вынесли в отдельное хранилище вложения. Это уже позволило несколько уменьшить число серверов для стораджей, но не сильно, так как вложения почти не влияли на расход IOPS. После того, как команда Виктора унесла со стораджей сами письма в Zepto-кластер, на стораджах остались только индексы. Их получилось уплотнить, и таким образом “старый” кластер похудел вполовину - осталось примерно 1.5К серверов. Уже хорошо, но задача стояла - уменьшить количество стораджей на порядок.

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

Хранилище для Почты