Привет! Я Антон, работаю в команде базовой инфраструктуры Контура и последние несколько лет занимаюсь развитием Kanso – распределенной системы хранения данных. Это наш форк GFS (Google File System). Мы развиваем Kanso уже более 15 лет, в статье я расскажу про это подробнее.

Kanso в проде сейчас – это 6 кластеров общим объемом физического хранения более 16PB. В самом крупном кластере 150к rps и 650 нод.

Зачем было нужно своё хранилище

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

Тогда наши инженеры обратили внимание на пейпер от компании Google, где была описана архитектура файлового хранилища. И решили на основе этого документа написать свою систему, которая удовлетворяла бы нашим потребностям. Так появилась Kanso.

Примеряем GFS

GFS – достаточно простая снаружи система. Она работает с файлами, основных операций над которыми всего две:

• Append(content) – дописать какой-то контент в конец файла. 

• Read(offset, length) – прочитать с какого-то оффсета часть данных. 

Эта довольно распространенная схема называется Append-only log или Write-ahead log. Например, данную идею можно встретить в журналировании в операционных системах.

Как и любая файловая система, GFS разделяет хранение метаданных и контента. За метаданные отвечает приложение Мастер – некий координатор и точка входа для клиента. А за контент отвечает чанк-сервер, его название я объясню немного позже. Можно мнить его себе как отдельный физический диск и приложение на нем.

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

Бьем файл на кусочки – чанки, раскладываем их по чанк-серверам (вот и смысл названия) и реплицируем с фактором 3. 

Чтобы найти файл, клиент идет к Мастеру как к координатору всего происходящего в системе. Получает ответ о местонахождении своих данных. Затем выполняет на соответствующем чанк-сервере чтение или запись. 

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

Наконец, перейдем к тому, как мы это всё начали писать и развивать. 

Важные этапы развития Kanso

2008: Начало. Стек хранилок

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

Да и не было задачи сделать полного клона – мы решали свои проблемы. И далее я буду описывать именно то, что мы сделали отлично от GFS. Если любопытно, оригинальный пейпер гугла можете прочесть по ссылке.

Итак, почти сразу после реализации первой версии MVP начали строить на основе Kanso стек хранилок в нашей компании. 

Берём простую советскую key-value структуру. Разбиваем диапазон всех ключей на отдельные рейнджи. Если отвечать на чтение из памяти, а данные и мутации над ними писать как операции Kanso файл, то таким образом можно построить распределенную in-memory таблицу. Мы называем её Zebra. Про одну из её подсистем я подробно рассказывал на одной из конференций DotNext. 

Zebra есть ни что иное как Big Table – ещё одна хранилка от Google.

Похожие на Big Table реализации можно увидеть и в open-source (например, YDB – Yandex Data Base). 

Почти сразу рядом с Zebra появилась и другая система – Echelon, распределенная очередь задач. Но самое большое распространение получило S3-like хранилище. Отмечу, что первый релиз Ceph состоялся лишь в 2012 году.

Представим, что есть задача хранить миллиарды разных бинарных контентов, произвольных. Это может быть видео, картинка. Всё, что угодно. Если создать на каждый такой контент отдельный файл в Kanso, то всё лопнет очень быстро. Поэтому создадим пул файлов, в который можно будет писать эти контенты. А для того, чтобы не потеряться, где и какие данные у нас в этой непрерывной бинарной «колбасе», нужно записать их координаты в таблицу. С этим прекрасно справится Zebra. Так у нас получился blob storage, который мы называем Drive (почти никак не связано с Райаном Гослингом :) ).

Если в общих чертах, то такой стек сейчас используется для хранения большинства данных в Контуре. 

2016: Kanso-3d

В 2016 году мы отказались от Kanso и написали с нуля новую систему – Kanso-3d (3 datacenters). Здесь можно сказать, что пути GFS и Kanso навсегда расходятся. Мы стали поддерживать полностью 3 ЦОД систему и гарантировать то, что все реплики находятся в разных физических местах. GFS же как-то распределяла их по своему кластеру (“spreading replicas across racks”). 

Также у нас появилась локальность на основе ЦОД-а. То есть клиент умеет очень быстро находить данные с реплики, которая рядом с ним находится. В GFS же предлагалась идея на основе IP-адресов. Мы ее не стали использовать, но взяли их изначальную идею про DNS alias-ы.

Для того, чтобы общаться с системой, клиенту нужно знать, где Мастер. У Google это было DNS имя, мы же написали статическую топологию. А уже на стороне сервера клиент получает либо редирект, либо ответ. Все эти манипуляции привели к тому, что мы получили гораздо более высокую отказоустойчивость, а еще снизилась latency и стало меньше служебного трафика. 

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

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

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

Данные клиента могли оказаться в любом кластере. В каком именно – мы не знаем. Поэтому нужно было отправить клиента в некого координатора. Этим координатором была некая мапа, у которой под капотом там LL(1) распознаватель. Обозначив распределение масок между кластерами, можно дать ему на вход файл и получить знание о местонахождении данных.

Благодаря этому мы скрыли от пользователей знание о том, что вообще есть какие-то кластера. Для него это просто Kanso.

Далее я подробно расскажу про кластер мастеров.

Кластер мастеров

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

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

За основу, по классике, был взят Zookeeper. А алгоритм представлял из себя какую-то вариацию многим известного raft. Он использовал несколько простых шагов: 

• блокировка на изменение состояния

• promise репликам

• synchronize с репликами

Где последний шаг относится к синхронизации состояния. Оно представляет из себя файл на диске у мастера. Мы называем его лог операций или оплог. Лидирующий мастер пишет туда операции, изменяющие систему (новый файл добавлен, файл удалён, файл переименован и так далее), но не сами пользовательские данные. Также он синхронно пишет операции остальным мастерам в их оплоги. Таким образом состояние реплицируется. Оплог после смены лидерства должен быть одинаковый у всех реплик и содержать максимально актуальные данные – значит нужно синхронизировать.

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

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

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

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

Решением стала асинхронная репликация. Такой механизм, когда резервисты сами, раз в какой-то период, подтягивают стейт из синхронных реплик или из лидера.

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

Далее, в 2024 году, проводя разные эксперименты по ускорению этого механизма, мы наткнулись на еще одну проблему. Есть сценарии, когда всё же требуется синхронизация большого файла состояния, но ранее описанные механизмы не спасают. Идею решения подсказала теория concurrency. А именно подход optimistic concurrency. Идея в том, чтобы не брать блокировку, и делать всё, что нужно. Только затем блокироваться и, удостоверившись что никто ничего не менял, выполнять оставшиеся действия. В нашем случае можно назвать это optimistic синхронизацией :) 

Таким образом, главные механизмы в кластере мастеров:

• Синхронная репликация

• Warm стейт

• Асинхронная репликация

• Минимум действий под блокировкой

2021: Unmanaged

Вернемся обратно в общую хронологию. В 2021 году мы встречаемся с особенностями дотнета, и на приложениях мастеров начинаются проблемы с памятью. 

Для понимания, что именно происходит, взглянем на приложение Мастера чуть глубже. 

Пример: клиент хочет читать файл abc.log по офсету 0, и хочет в результате 1 байт. Что требуется мастеру для ответа? 

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

Отвечать клиенту мы будем списком реплик. Чтобы знать этот набор, нужно понимать, как чанки раскладываются по чанк-серверам. Для этого ещё одна структура. Далее клиент пойдёт в какую-то реплику и прочтёт свой байт.

Эти мапы хранят огромное количество dotnet объектов. И, что самое главное, они крайне долгоживущие. Никто их никогда не удалит, ведь это суть стейта. Следствием отсюда является адский stop the world GC второго поколения. Иногда это даже приводило к протуханию сессии к zookeeper. А значит начинался процесс со сменой лидерства, который, даже мог зациклиться. 

Как мы это решили? Ответ уже есть в заголовке. Мы написали свою Unmanaged обвязку для самых горячих объектов в системе. По большей части, это коснулось тех самых гигантских мапов: появились UnmanagedHashSet, UnmanagedDictionary. На добавку и Unmanaged Buffer pools для внутренних протоколов.

Решение опасное, но работает.

2022: Не храним ненужное

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

Мало того, что мы сэкономили в моменте достаточное количество объема. На текущий момент, если бы мы не занимались этим «откусыванием», было бы +2 петабайта данных. Более важно, что предотвратили бесконечный рост в некоторых сценариях. Когда, например, у Big Table в таблице хранится 1 ключ, а история его пача занимает терабайты.

2022: Хитрый update кластера

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

Всё потому, что чанк-сервер хранит чанки как файлы на диске, то есть как открытый файловый хэндл. Тогда ещё всё работало на windows , и чтобы выключить приложение, нужно от всех них избавиться. Остановить при релизе старое приложение и закрыть миллионы хэндлов порой занимало 10-15 минут. 

Решение нашлось на уровне операционной системы. Работало это так.

• Запускался служебный процесс, которому с помощью WinAPI функций передавались все файловые хэндлы. По сути, мы их не закрывали, а просто копировали. 

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

• Далее от служебного приложения хэндлы передавались обратно. 

Таким образом, используя знания уровня операционной системы, мы решили проблему, которая возможно и не решается на уровне рантайма. Я не знаю, как можно в дотнет приложении ускорить закрытие файловых хэндлов. Пишите в комментарии, если знаете, как это можно сделать!

2023: Учения

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

«Ты не отказоустойчив, пока не проверил это!» 

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

Дополнительно это кладезь интересных задач и вызовов, которых порой в долгоиграющем проекте становится всё меньше

2024: Linux + Docker + on-prem

В 2024 году компания явно зафиксировала желание продавать продукты, которые будут запускаться на чужих площадках. И от нас, как от хранителей стека, потребовалось быстро к этому приспособиться. Мы переехали на Linux и запустились в Docker-е. И сейчас умеем масштабироваться так же, как и любые взрослые распределенные системы. 

Что осталось за кадром

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

Откликнитесь в комментариях, если что-то из перечисленного вам интересно. Возможно будет ещё одна статья.

Заглянем в будущее

Что нас ждет дальше? С какими вызовами столкнется команда хранения данных? 

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

Масштабируемость

При увеличении скорости появления новых кластеров, шардировать их становится неудобно. Смена маппинга и перевод данных – долго и дорого, ведь этим занимается инженер. Хотелось бы просто добавлять серверов для данных в систему, и чтобы оно само развозилось. У нас есть идея, как это сделать. Но я предлагаю посмотреть, как это сделали в Google, развивая GFS.

Как вы уже знаете, на основе GFS можно сделать Big Table. А теперь давайте в этот Big Table положим те гигантские мапы, которые мы переводили в Unmanaged. То есть вынесем их во внешний сервис. Осталось добавить серверов с данными. И что у нас получается? Это примерная архитектура Google Colossus (CFS) – развитие Google File System, к которому ребята из гугла пришли ещё в 2009 году. 

CFS – это распределённая файловая система. Там можно хранить append-only файлы. А значит поверх можно сделать Big Table. Догадываетесь, к чему я? Добавляем серверов с данными. И мы получим Colossus 2 поколения (или второго этажа)!

Благодаря такой матрёшке, Google решает проблему большого объема метаданных. Они пишут: 

“If we host a Colossus on Colossus

100PB data -> 10TB metadata 

10TB metadata -> 1GB metadata

1GB metadata -> 100KB data”

Эффективное хранение

Репликация, которая есть у нас сейчас – не самое эффективное решение хранения большого объема данных. Мировой опыт подсказывает использовать здесь Erasure codes. Упоминание об этом можно увидеть в Colossus, в RedHat Ceph.

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

Снова настоящее

И всё же вернёмся в текущий момент. А там Kanso всё еще 6 кластеров общим объемом физического хранения более 16PB.

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