Elliptics от Яндекса. Как с его помощью создать своё отказоустойчивое хранилище
Добрый день, дорогие читатели! В прошлых статьях я в общих чертах рассказывал про наше открытое отказоустойчивое хранилище Elliptics, а также немного опускался в детали. Сегодня же я вам наглядно расскажу и покажу, как использовать Elliptics на примере создания своей собственной отказоустойчивой ХабраМузыки.
ХабраМузыка — это ваше личное хранилище музыки с поддержкой региональности, реплицирования данных, минимальной нагрузкой на диск и сеть, а также простым HTTP API, который можно использовать в любом вашем приложении или на личном сайте.
Под катом — пошаговые подробности.
Первым делом нам нужно поставить Elliptics. Для этого мы собираем пакеты в нашем репозитории под такие распространненные серверные системы, как: Debian Wheezy; Ubuntu Precise 12.04; Ubuntu Trusty 14.04; RedHat 6 / CentOS 6. Для других дистрибутивов вы можете собрать систему самостоятельно из исходников, которые расположены на нашем GitHub-аккаунте.Для дальнейшей работы необходимо поставить пакеты:
$ sudo apt-get install elliptics elliptics-client rift или $ sudo yum install elliptics elliptics-client rift Если у вас есть желание попробовать в деле API для C++, то также следует поставить пакет elliptics-dev или elliptics-client-devel в зависимости от дистрибутива.
Отлично, пакеты поставлены. Теперь пришла пора запускать свой кластер. В рамках данного туториала мы будем запускать все сервера на одной машине. В реальной же ситуации их стоит разносить на разные машины и дата-центры. На GitHub расположены все конфигурационные файлы, которые понадобятся для нашего туториала.
Для начала ознакомимся с примером конфигурационного файла node-1.json:
{ «loggers»: { «type»:»/srv/elliptics/1/log.txt», «level»: 4 }, «options»: { «join»: true, «flags»: 4, «remote»: [ «autodiscovery:224.0.0.5:1025:2» ], «address»: [ «localhost:1025:2» ], «wait_timeout»: 60, «check_timeout»: 60, «io_thread_num»: 16, «nonblocking_io_thread_num»: 16, «net_thread_num»: 4, «daemon»: true, «auth_cookie»:»<-Don't forget to change it->», «cache»: { «size»: 1073741824 }, «indexes_shard_count»: 16, «monitor_port»: 20000 }, «backends»: [ { «type»: «blob», «group»: 1, «history»:»/srv/elliptics/1/history/», «data»:»/srv/elliptics/1/blob/data», «sync»:»-1», «blob_flags»:»129», «blob_size»:»10G», «records_in_blob»:»1000000» } ] } Чтобы запустить Elliptics с этим конфигурационным файлом, надо создать следующие каталоги:/srv/elliptics/1/ — здесь в файл log.txt будут писаться логи с уровнем Debug; /srv/elliptics/1/blob/ — здесь будут лежать блобы с данными; /srv/elliptics/1/history/ — здесь будет лежать файлик с началами интервалов из DHT-кольца. Подробно об остальных указанных и неуказанных полях можно прочитать в нашей документации.Для запуска достаточно выполнить:
$ dnet_ioserv -c node-1.json Но мы ведь хотим построить отказоустойчивый кластер. Для этого надо поднять еще хотя бы одну машину во второй группе. В её конфиге нужно будет указать другой адрес, а также другие remote-адреса. В итоге, пока используются конфигурационные файлы node-1.json, node-2.json для хранения данных в 2 копиях.Создадим каталоги, нужные для первых двух нод, и запустим каждую из них:
$ for id in `seq 1 2`; do mkdir /srv/elliptics/$id/{history, blob} -p; done
$ for id in `seq 1 2`; do dnet_ioserv -c node-$id.json; done
А теперь, когда сервера у нас запущены, приступим к знакомству с низкоуровневыми C++ и Python API. Для этого каждое действие я буду показывать сразу на двух языках с пояснениями.Подключение к серверам
Сперва нужно создать клиентскую ноду для подключения к серверам Elliptics.С++:
#include
using namespace ioremap: elliptics;
file_logger logger (»/dev/stderr», DNET_LOG_ERROR); node n (file_logger);
n.add_remote («localhost», 1025); n.add_remote («localhost», 1026);
session sess (n); sess.set_groups ({ 1, 2, 3 }); Python: import elliptics log = elliptics.Logger (»/dev/stderr», elliptics.log_level.debug) node = elliptics.Node (log)
node.add_remote («localhost», 1025) node.add_remote («localhost», 1026)
sess = elliptics.Session (node) sess.group = [ 1, 2, 3 ] Записываем данные в Elliptics Запишем в Elliptics немного данных: C++:
auto async_result = sess.write_data (std: string («test_key»), «some data», offset); Python: async_result = sess.write_data («test_key», «some_data», offset) Вышеприведенный код отправит запрос записи данных на сервера Elliptics. Однако после этого нужно дождаться ответов и обработать их. Многие команды посылают по несколько запросов на сервера Elliptics«а. Так, например, запись данных шлет запросы одновременно в каждую группу по запросу. На каждый запрос приходит ответ, из которого можно узнать как некоторые общие для всех команд сведения (например, адрес сервера, обработавшего запрос, или код результата выполнения команды), так и специфичную для данной команды информацию (такую, как путь к блобу, в который была произведена запись, с координатами внутри него).Обработка ответа от сервера Существует несколько вариантов обработки результата выполнения команды. Их стоит рассмотреть подробнее.Синхронный запрос C++: try { async_result.wait (); } catch (error &e) { std: cerr << "Error while write execution: " << e.message(); return; } auto sync_result = async_result.get(); for (write_result_entry entry : async_result.get()) { // Process entry std::cerr << "Received answer from: " << dnet_server_convert_dnet_addr(result.address()) << "status: " << result.status() << std::endl; } Python: try: async_result.wait() except Exception as e: print "Write failed: ", e
for entry in async_result.get (): print («Reply from {}, status: {}».format (entry.address, entry.error.code)) От кидания исключений можно избавиться, вызвав у session метод set_exceptions_policy, который присутствует как в C++, так и в Python API. В таком случае нужно делать ручную проверку на ошибки: C++:
if (async_result.error ()) { // Логика обработки ошибки } Python: if (async_result.error) { // Логика обработки ошибки } Итерируемый запрос Иногда суммарный объем данных от сервера может быть слишком большим, чтобы разом поместить его в память. Также может иметься требование обработки данных по мере их поступления. В таком случае можно воспользоваться итерированием: тело цикла будет выполняться на каждый ответ, пришедший от сервера, по мере их выступлений. Между пакетами поток пользователя будет спать.C++:
for (auto entry: async_result) { } Python: for entry in async_result: pass Асинхронный запрос Также, как только ответ будет полностью получен, можно выполнить свой код в потоке Elliptics«а.C++:
async_result.connect ([] (const sync_write_result &result, const error_info &error) { if (error) { // Обработка ошибки return; } for (auto entry: result) { // Обработка ответа } }); Python:
def handler (results, error): if error.code!= 0: // Обработка ошибки return for entry in results: // Обработка ответа pass async_result.connect (handler) Асинхронный итеративный запрос Этот способ рекомендуется использовать, если данных от сервера ожидается много, их хочется обрабатывать по мере поступления и не хочется блокировать ни один из потоков для ожидания данных. Первый callback будет вызываться на каждый полученный от сервера ответ, а последний вызовется после получения последнего ответа. Причем единственным аргументом будет информация о возможной ошибке, если итоговый код ответа не нулевой.C++
async_result.connect ([] (const write_result_entry &entry) { // Обработка ответа }, [] (const error_info &error) { if (error) { // Обработка ошибки } }); Python: def result_handler (result, error): // Обработка ответа pass def error_handler (results, error): if error.code!= 0: // Обработка ошибки pass async_result.connect (result_handler, error_handler) Чтение данных Для чтения данных достаточно выполнить нижеприведенный код, а затем обработать ответ так же, как в случае write«а: C++:
auto async_result = sess.read_data (std: string («test_key»), offset, size); Python: async_result = sess.read_data («test_key», offset, size) При этом стоит помнить, что на чтение сервер мог не найти файл с первого раза. В таком случае количество ответов может быть более одного, причем все, кроме последнего, будут с отрицательным кодом ошибки.Удаление данных Удалять данные также просто.C++:
auto async_result = sess.remove_data (std: string («test_key»));
Python:
async_result = sess.remove_data («test_key»)
И многое другое
доступно в нашей документации.
В качестве примера рассмотрим способ создания своего сервиса по прослушиванию музыки.
Для начала нам нужно поднять еще как минимум один сервер Elliptics для хранения такой мета-информации Rift«а, как бакеты.
$ dnet_ioserv -c node-3.json Теперь надо настроить и поднять Rift, его конфиг приведен в GitHub-репозитории под именем rift.json, рассмотрим его поближе: { «endpoints»: [ »0.0.0.0:8080» ], «backlog»: 128, «threads»: 2, «buffer_size»: 65536, «logger»: { «file»:»/srv/rift/log.txt», «level»: 4 }, «daemon»: { «fork»: true, «uid»: 1000 }, «monitor-port»: 21000, «application»: { «remotes»: [ «localhost:1025:2», «localhost:1026:2», «localhost:1027:2» ], «groups»: [], «metadata-groups»: [ 3 ], «bucket»: { «timeout»: 60 }, «read-timeout»: 10, «write-timeout»: 16, «redirect-port»: 8081, «path-prefix»:»/srv/elliptics/» } } Конфигурационный файл можно разбить на две секции. Первая отвечает за конфигурирование HTTP-сервера в целом: «endpoints» — список портов и unix-сокетов, которые слушает сервер на входящие соединения; «logger» — конфигурация системы логгирования; «daemon» — описание демонизации сервера. Вторая секция — «application» — отвечает за конфигурацию Rift«а: «remotes» — список узлов Elliptics«а; «metadata-groups» — список групп, в которых хранится вся мета-информация, например, бакеты; «redirect-port» — порт, который будет слушать Nginx на машинах с данными; «path-prefix» — начальная часть пути к блобам на машинах с данными. Подробно обо всех параметрах конфигурационного файла можно прочитать здесь. $ rift_server -c rift.json Все операции в Rift с активированными бакетами требуют авторизации. Алгоритм формирования подписи описан в документации. Здесь же я буду использовать крайне простой http-клиент на Python, который все запросы будет подписывать надлежащим образом. Он лежит в архиве и называется http_auth.py.Создадим бакет, в который мы будем заливать нашу музыку. Причем хотелось бы, чтобы правом залива музыки обладал только администратор ресурса, то есть мы, а слушать мог кто угодно. Рассмотрим файл конфигурации бакета:
{ «groups»: [ 1, 2 ], «acl»: [ { «user»: «admin», «token»: «favorite_music», «flags»: 6 }, { «user»:»*», «token»: «very_secret_token», «flags»: 1 } ] } Первое поле «groups» говорит, что данные будут писаться в двух копиях, по копии в группах 1 и 2. Подробнее о группах можно прочитать в предыдущих статьях.Второе поле «acl» содержит права доступа для различных пользователей, в данном примере для каждого пользователя были указаны 3 поля: «user» — публичный id пользователя; «token» — секретный токен, по которому происходит авторизация; «flags» — описание прав пользователя, на данный момент доступны следующие флаги и их битовые комбинации:1 — для данного пользователя не требуется авторизация; 2 — данный пользователь имеет право для записи данных в бакет; 4 — данный пользователь имеет право для изменения данных самого бакета (список групп, acl и т.д.). Правами для записи новых файлов и изменения бакета обладает только «admin», так же указан его секретный ключ, с помощью которого будет происходить авторизация. А вот правом чтения обладает любой внешний пользователь без необходимости авторизации. Создадим теперь бакет music в директории all: $ ./http_auth.py http://localhost:8080/update-bucket/all/music --file update-bucket.json 200 Директория — это особенная сущность, которая позволяет, например, найти все содержащиеся в ней бакеты. В созданный бакет зальем немного музыки по ключу example.mp3: $ ./http_auth.py http://localhost:8080/upload/music/example.mp3? user=admin --file example.mp3 --token favorite_music 200 После этого можно слушать музыку прямо из любого проигрывателя: $ mplayer http://localhost:8080/get/music/example.mp3 Ручка /get поддерживает заголовки Last-Modified и Range для более эффективного управления траффиком, но это все равно не самое рациональное использование ресурсов. Гораздо эффективнее было бы стримить данные с помощью Nginx«а с ближайших к пользователю физических машин, на которых данный файл хранится. Это достигается за счет поддержки региональности в Elliptics.Для этого нам нужен Nginx с нашим eblob-модулем. Его можно найти на GitHub«е.
Собрать его можно с использованием любого подходящего Nginx«а (поддерживаются версии по 1.7.х) примерно следующим образом :
$ git clone https://github.com/toshic/nginx_patches.git $ wget 'http://nginx.org/download/nginx-1.7.0.tar.gz' $ tar -xzvf nginx-1.7.0.tar.gz $ cd nginx-1.7.0 $ ./configure --prefix=/opt/nginx --add-module=…/nginx_patches/nginx-eblob $ make $ sudo checkinstall # :) Для активации eblob-модуля достаточно добавить блок «eblob;» в location-секцию. Таким образом, минимальный конфигурационный файл выглядит так: worker_processes 8; events { worker_connections 1024; } http { sendfile on; server { listen 8081; server_name localhost; location / { root /var/srw; eblob; } } } Разумеется, для боевого окружения Nginx следует настраивать не настолько наивно.Отлично, теперь, когда nginx настроен и запущен, можно стримить музыку напрямую с него:
$ mplayer -q http://localhost:8080/redirect/music/example.mp3 По данному запросу сервер сгенерирует url специального вида, получив который Nginx сможет отдавать музыку прямо с blob«ов, в который Elliptics хранит свои данные.Гораздо подробнее про Rift можно почитать в нашей документации. Там же имеется подробное описание всех доступных методов.
Про это и многое другое можно почитать в нашей документации на doc.reverbrain.com: Напоминаю, что все предоставленные в данной статье проекты открыты и расположены на GitHub: https://github.com/reverbrain.
