Как быстро собрать метаданные сотен миллиардов объектов в S3

Привет! Это Артём Мурашко, разработчик Cloud Storage Services в Yandex Infrastructure. Уже четвёртый год я работаю с объектным хранилищем Яндекса. В этой статье расскажу, как мы выстраивали систему сбора метаданных сотен миллиардов объектов в S3, с какими сложностями столкнулись и как нам удаётся листить миллиарды объектов за часы.

Архитектура Yandex S3Архитектура Yandex S3

Что представляет собой Yandex Object Storage? Это масштабируемое S3-хранилище для данных любого типа, файлов и архивов. Сервис автоматически создаёт несколько копий каждого объекта, размещает их в разных зонах доступности, обеспечивает резервное копирование и восстановление данных.

Для хранения данных пользователи могут создавать бакеты (именованные логические контейнеры) и управлять ими с помощью консоли Yandex Cloud. API Yandex Object Storage совместим с API Amazon S3, поэтому можно использовать инструменты, предназначенные для работы с S3.

Когда пользователь пишет объект, он попадает в его бакет. У каждого объекта есть данные и метаданные:

• данные хранятся в нашем низкоуровневом хранилище с простым интерфейсом — пользователи отправляют байты, он отдаёт им ключ (по ключу можно как прочитать данные, так и удалить их);
• метаданные хранятся в PostgreSQL — они задают ключ от низкоуровнего хранилища, имя объекта, размер, дату записи.

Особенность S3 — в том, что файлы хранятся плоским списком: это позволяет масштабировать хранилище.

У нас предусмотрены три сценария работы с S3:

• Внутренняя инсталляция — для сервисов и команд Яндекса.
• Облачная инсталляция (Yandex Cloud) — для организаций, которым нужно гибкое управление сервисами и масштабирование под нагрузку.
• On-premises — для компаний, которым требуется полный контроль над данными и управление процессами внутри собственной инфраструктуры. Обычно у таких организаций есть особые требования к производительности и соответствию законодательным нормам.

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

Остановимся подробнее на метаданных.

Рассмотрим следующую схему — это путь клиентского запроса к S3:

Как работает система? Запрос попадает на балансер, тот проксирует на один из наших инстансов бэкенда, который обслуживает S3-протокол. Далее, в зависимости от сценария использования, мы идем в PostgreSQL и берём оттуда метаданные. Прочитали ключ, сходили с ним в хранилище, прочитали данные, проксировали обратно пользователю. И наоборот: записали данные, получили ключ, записали в PostgreSQL метаданные.

Объектов и метаданных много, и все они помещаются на один хост PostgreSQL — нужно шардироваться. Пространство хешей ключей бакета нарезается на непересекающиеся диапазоны — чанки. Каждый объект имеет ключ, hash (key) попадает в один диапазон — это его чанк. Чанки живут на разных шардах: чанков у бакета может быть больше одного, как и шардов у бакета. Подробнее о шардировании читайте в статье Павла Левдика на Хабре.

Отмечу также типы баз данных для хранения метаинформации об объектах:

• S3meta — в ней хранятся бакеты, их настройки, адресация чанков, информация о пользователях, статистика;
• S3db — здесь лежат метаданные объектов, информация о чанках на шарде.

Проблема листинговПроблема листингов

Достаточно часто пользователям нужно оптимизировать хранение метаданных объектов и провести аудит. Как это сделать быстро и эффективно?

Если говорить о листинге объектов, то в спецификации S3 есть методы ListObjects и ListObjectsVersions (нужен для листинга версий объектов в случае включенного версионирования в бакете). Запрос простой, он возвращает до 1000 ключей, отсортированных в лексикографическом порядке, и принимает ContinuationToken. Мы можем с пагинацией обойти все объекты бакета и достать метаданные.

Давайте посмотрим на листинг объектов со стороны пользователя:

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

Теперь посмотрим на тот же листинг, но со стороны S3:

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

Но мы не можем просто так сходить на один шард и достать первую тысячу отсортированных ключей. Получаем поход на каждый шард бакета: листим на каждом шарде, потом в Runtime всё это мёржим, сортируем, возвращаем первую тысячу — и так, пока не обойдём весь бакет. Если взять среднее время ответа листов — 50 миллисекунд — то, например, бакет с 10 миллиардами объектов нужно будет обходить 6 дней, а это долго.

Таким образом, при листинге пользователи могут столкнуться с несколькими негативными факторами:

• пагинация — надо писать логику сбора данных;
• долгое выполнение — ждать часы;
• ограниченный набор метаданных — только Key, Size, LastModified, ETag, StorageClass, Owner, VersionId (нет ACL объектов, информации по шифрованию и других метаданных);
• необходимость решать вопросы с инфраструктурой (запросы нужно откуда-то делать, а промежуточные данные где-то хранить).

Для нашего сервиса тоже есть сложности:

• мы не контролируем нагрузку — чем больше у бакета шардов, тем больше запросов он делает на каждый из объектов;
• проблема «шумных соседей» — чем больше бакетов, тем больше вероятность наткнуться на деградированный шард и снизить скорость ответа.

Как эту проблему решает Amazon? Там придумали механизм S3 Inventory:

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

Конфигурация задаётся так:

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

S3 Inventory: реализацияS3 Inventory: реализация

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

Сбор метаданныхСбор метаданных

Самый сложный вопрос — сбор метаданных. Листинг можно делать несколькими способами:

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

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

Как выглядит обработка группы?

Идём по группе слева направо, объединяем чанки, следующие друг за другом и лежащие на одном шарде, в подгруппу:

Проверяем миграции и копируем подгруппу:

Скопировалось порядка 90 000 строк, но всё это заняло 7,6 секунд. Почему не меньше?

Смотрим на shared read — 62 000. Когда мы фильтруем по границам подгруппы, нет гарантии, что все объекты, лежащие в подгруппе, находятся на одной подгрупповой странице. Соответственно, чтобы залить все объекты, нужно потенциально достать очень много страниц как из кеша, так и с диска.

Ещё один вопрос: как копировать? Например, можно прямо в Runtime нашего бэкенда селектить и обрабатывать метаданные, а затем выгружать. Однако в PostgreSQL есть функция копирования. Она принимает в себя SQL-запрос и непрерывным потоком отдаёт в stdout то, что получает из селекта.

В этом примере функция сама выдаёт CSV-формат (остаётся перенаправить этот поток в нужный бакет):

Какой из способов лучшеКакой из способов лучше

Мы проанализировали все три способа сбора метаданных, изучив их плюсы и минусы:

Как мы и предполагали, пошардовый обход — самый быстрый, в нём идёт фильтрация только по bucket_id. Но одна из проблем метода — дубликаты или потери. Если чанки в самом начале делить на отрезки и не модифицировать их, тогда чанк будет разрастаться по количеству объектов. Нам же важна возможность балансировать нагрузку по шардам, а для этого нужно поддерживать оптимальный размер чанка. Задачу решает фоновый процесс, который находит большие чанки и делит их в определённой пропорции.

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

Но может быть и обратная ситуация, например, мы не долистим какой-то чанк (он переедет в моменте перехода на следующий шард). Транзакции Repeatable Read не спасут ситуацию. Синхронизации между шардами в этом случае нет.

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

Есть нюансы и по транзакциям. При пошардовом обходе мы берём repeatable-транзакцию и потом фильтруем по bucket_id. Объектов на шарде может быть много, а значит, транзакция будет долгой. В почанковом и групповом обходах транзакции достаточно короткие, поэтому листить можно быстрее.

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

Что показали бенчмарки?Что показали бенчмарки?

Каждый из способов сбора метаданных мы тестировали с одним и тем же потоком. Например, у нас был бакет-миллиардник (№ 4 на скрине) — обход по шардам справился за 29 часов, по чанкам — за 61 час, а по группам — за 54 часа:

Время обхода по чанкам и группам примерно равное, но чанков у бакета много, поэтому генерируется почти 100 000 задач на обработку. Пошардовый быстрее примерно в два раза: генерируется всего 4 задачи (бакет живёт на четырёх шардах), но здесь возможны дубликаты.

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

S3 Inventory: умный обходS3 Inventory: умный обход

Но как сделать групповой обход ещё умнее? Здесь важно учитывать, что в каждом бакете есть неизменные объекты, например, логи (записали один раз, и они лежат там 5 лет, их не нужно обходить каждый день). В этом случае нужно полистить весь бакет и каждый день считать изменения тех объектов, которые были модифицированы, добавлены или удалены. И потом всё это собрать:

Мы хотели удешевить решение, но MapReduce — это достаточно дорого, его нужно поддерживать, включая on-premises-инсталляции.

Как выглядит наш пайплайн обработки конфигурации?

Введём понятие job — это сущность, которая соответствует паре: конфигурация на бакете + дата запуска инвентаря. Нужно скопировать объекты, потом переложить их в клиентский бакет, сформировать манифест. Когда в бакете не хватает квоты на добавление новых объектов, мы фейлим job и больше туда ничего не пишем.

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

Зачем нужны два состояния — Copy и Transfer?Зачем нужны два состояния — Copy и Transfer?

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

Кто обрабатывает эту машину?

• Scheduler — обнаруживает конфигурации, требующие обработки, и создаёт соответствующие jobs.
• Manager — осуществляет переходы между состояниями, формирует задачи для обработки.
• Worker — исполнитель задач; каждый worker специализируется на одном состоянии и является единицей параллелизма. Workers помогают контролировать нагрузку.

Пример того, как это может работать:

Результаты запускаРезультаты запуска

В таблице приведено сравнение S3 Inventory и ListObjects по времени обработки для крупных бакетов (если взять 50 миллисекунд как время получения ответа):

Нам довелось обработать 80-миллиардный бакет за 22 часа: при последовательном листинге на это ушло бы 46 дней. С момента запуска у нас ежедневно обрабатывается более 100 конфигураций. Такая схема для крупных бакетов в десятки раз быстрее, чем ListObjects.

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

Какие плюсы дал наш подходКакие плюсы дал наш подход

• Автономность: пользователям не нужно писать свою логику обхода, не нужна своя инфраструктура — всё есть на нашей стороне.
• Скорость: для больших бакетов процесс идёт быстрее, что позволяет экономить сотни часов (в том числе и нам). Пользователи получают высокую скорость, а мы — предсказуемую нагрузку на шарды.
• Оптимизация: снижает нагрузку на Control Plane (мы умеем её контролировать). Система даёт 15+ полей метаданных, а выгрузка готова к аналитике.

Подход эффективный, но стоит учитывать, что для маленьких бакетов такой механизм избыточен. Если бакет хранит 10 000 объектов, то эффективнее руками сделать 10 запросов ListObjects, а не ждать выгрузки. Стоит помнить и о времени получения отчётов: мы запускаемся каждую ночь — нужно будет дождаться следующего дня, чтобы поработать с конфигурацией.

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

Отчёт всегда включает поля:

• BUCKET_NAME — имя исходного бакета;
• KEY — ключ объекта.

При выгрузке по всем версиям объектов также добавляются:

• VERSION_ID — идентификатор версии;
• IS_LATEST — флаг последней версии;
• DELETE_MARKER — флаг маркера удаления.

А это опциональные метаданные, которые можно указать в конфигурации:

• SIZE — размер в байтах, кроме размера незавершенных частей составных загрузок, метаданных объекта и маркеров удаления;
• LAST_MODIFIED_DATE — дата создания или последнего изменения;
• ETAG — хеш;
• STORAGE_CLASS — класс хранилища;
• INTELLIGENT_TIERING_ACCESS_TIER — уровень доступа объекта в умном хранилище;
• IS_MULTIPART_UPLOADED — маркер составной загрузки;
• ENCRYPTION_STATUS — статус шифрования;
• OBJECT_LOCK_RETAIN_UNTIL_DATE — дата окончания блокировки версии;
• OBJECT_LOCK_MODE — тип блокировки версии;
• OBJECT_LOCK_LEGAL_HOLD_STATUS — статус бессрочной блокировки версии;
• CHECKSUM_ALGORITHM — алгоритм, используемый для расчёта контрольной суммы;
• OBJECT_ACCESS_CONTROL_LIST — ACL в кодировке base64;
• OBJECT_OWNER — идентификатор аккаунта владельца.

Актуальный список доступных параметров смотрите в Справочнике API.