2020-highload-dht

Курсовой проект 2020 года курса "Highload системы" в Технополис.

Этап 1. HTTP + storage (deadline 2020-09-30)

Fork

Форкните проект, склонируйте и добавьте upstream:

$ git clone git@github.com:<username>/2020-highload-dht.git
Cloning into '2020-highload-dht'...
...
$ git remote add upstream git@github.com:polis-mail-ru/2020-highload-dht.git
$ git fetch upstream
From github.com:polis-mail-ru/2020-highload-dht
 * [new branch]      master     -> upstream/master

Make

Так можно запустить тесты:

$ ./gradlew test

А вот так -- сервер:

$ ./gradlew run

Develop

Откройте в IDE -- IntelliJ IDEA Community Edition нам будет достаточно.

ВНИМАНИЕ! При запуске тестов или сервера в IDE необходимо передавать Java опцию -Xmx256m.

В своём Java package ru.mail.polis.service.<username> реализуйте интерфейс Service и поддержите следующий HTTP REST API протокол:

• HTTP GET /v0/entity?id=<ID> -- получить данные по ключу <ID>. Возвращает 200 OK и данные или 404 Not Found.
• HTTP PUT /v0/entity?id=<ID> -- создать/перезаписать (upsert) данные по ключу <ID>. Возвращает 201 Created.
• HTTP DELETE /v0/entity?id=<ID> -- удалить данные по ключу <ID>. Возвращает 202 Accepted.

Возвращайте реализацию интерфейса в ServiceFactory.

Реализацию DAO берём из весеннего курса 2020-db-lsm, либо запиливаем adapter к уже готовой реализации LSM с биндингами на Java (например, RocksDB, LevelDB или любой другой).

Проведите нагрузочное тестирование с помощью wrk в одно соединение:

• PUT запросами на стабильной нагрузке (wrk должен обеспечивать заданный с помощью -R rate запросов)
• GET запросами на стабильной нагрузке по наполненной БД

Почему не curl/F5, можно узнать здесь и здесь.

Приложите полученный консольный вывод wrk для обоих видов нагрузки.

Отпрофилируйте приложение (CPU и alloc) под PUT и GET нагрузкой с помощью async-profiler. Приложите SVG-файлы FlameGraph cpu/alloc для PUT/GET нагрузки.

Объясните результаты нагрузочного тестирования и профилирования и приложите текстовый отчёт (в Markdown).

Продолжайте запускать тесты и исправлять ошибки, не забывая подтягивать новые тесты и фиксы из upstream. Если заметите ошибку в upstream, заводите баг и присылайте pull request ;)

Report

Когда всё будет готово, присылайте pull request со своей реализацией и оптимизациями на review. Не забывайте отвечать на комментарии в PR (в том числе автоматизированные) и исправлять замечания!

Этап 2. Многопоточность (deadline 2020-10-07)

Обеспечьте потокобезопасность реализации DAO с использованием примитивов java.util.concurrent.*. Прокачаться можно с руководством Java Concurrency in Practice.

Реализуйте асинхронный flush MemTable в SSTable, чтобы не блокировать пользовательские запросы.

Сконфигурируйте HTTP сервер, чтобы он обрабатывал запросы с помощью пула из нескольких потоков.

Проведите нагрузочное тестирование с помощью wrk в несколько соединений (не меньше 16):

• PUT запросами на стабильной нагрузке (wrk должен обеспечивать заданный с помощью -R rate запросов)
• GET запросами на стабильной нагрузке по наполненной БД

Приложите полученный консольный вывод wrk для обоих видов нагрузки.

Отпрофилируйте приложение (CPU, alloc и lock) под PUT и GET нагрузкой с помощью async-profiler. Приложите SVG-файлы FlameGraph cpu/alloc/lock для PUT/GET нагрузки.

Объясните результаты нагрузочного тестирования и профилирования и приложите текстовый отчёт (в Markdown).

Report

Когда всё будет готово, присылайте pull request со своей реализацией и проведёнными оптимизациями на review.

Этап 3. Асинхронный сервер (deadline 2020-10-14 00:00:00 MSK)

Вынесите обработку запросов в отдельный ExecutorService с ограниченной очередью, чтобы разгрузить SelectorThreadы HTTP сервера.

Проведите нагрузочное тестирование с помощью wrk2 с большим количеством соединений (не меньше 64) PUT и GET запросами.

Отпрофилируйте приложение (CPU, alloc и lock) под PUT и GET нагрузкой с помощью async-profiler.

Report

Когда всё будет готово, присылайте pull request с изменениями, результатами нагрузочного тестирования и профилирования, а также анализом результатов по сравнению с предыдущей (синхронной) версией.

Этап 4. Шардирование (deadline 2020-10-21 00:00:00 MSK)

Реализуем горизонтальное масштабирование через поддержку кластерных конфигураций, состоящих из нескольких узлов, взаимодействующих друг с другом через реализованный HTTP API. Для этого в ServiceFactory передаётся статическая "топология", представленная в виде множества координат всех узлов кластера в формате http://<host>:<port>.

gradle run теперь стартует Cluster из трёх нод.

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

Реализуйте один из алгоритмов распределения данных между узлами, например, consistent hashing или rendezvous hashing.

Report

Присылайте pull request со своей реализацией поддержки кластерной конфигурации на review. Не забудьте нагрузить, отпрофилировать и проанализировать результаты профилирования под нагрузкой. С учётом шардирования набор тестов расширяется, поэтому не забывайте подмёрдживать upstream.

Этап 5. Репликация (deadline 2020-10-28 00:00:00 MSK)

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

HTTP API расширяется query-параметром replicas, содержащим количество узлов, которые должны подтвердить операцию, чтобы она считалась выполненной успешно. Значение параметра replicas указывается в формате ack/from, где:

• ack -- сколько ответов нужно получить
• from -- от какого количества узлов

Таким образом, теперь узлы должны поддерживать расширенный протокол (совместимый с предыдущей версией):

• HTTP GET /v0/entity?id=<ID>[&replicas=ack/from] -- получить данные по ключу <ID>. Возвращает:

  • 200 OK и данные, если ответили хотя бы ack из from реплик
  • 404 Not Found, если ни одна из ack реплик, вернувших ответ, не содержит данные (либо самая свежая версия среди ack ответов -- это tombstone)
  • 504 Not Enough Replicas, если не получили 200/404 от ack реплик из всего множества from реплик

• HTTP PUT /v0/entity?id=<ID>[&replicas=ack/from] -- создать/перезаписать (upsert) данные по ключу <ID>. Возвращает:

  • 201 Created, если хотя бы ack из from реплик подтвердили операцию
  • 504 Not Enough Replicas, если не набралось ack подтверждений из всего множества from реплик

• HTTP DELETE /v0/entity?id=<ID>[&replicas=ack/from] -- удалить данные по ключу <ID>. Возвращает:

  • 202 Accepted, если хотя бы ack из from реплик подтвердили операцию
  • 504 Not Enough Replicas, если не набралось ack подтверждений из всего множества from реплик

Если параметр replicas не указан, то в качестве ack используется значение по умолчанию, равное кворуму от количества узлов в кластере, а from равен общему количеству узлов в кластере, например:

• 1/1 для кластера из одного узла
• 2/2 для кластера из двух узлов
• 2/3 для кластера из трёх узлов
• 3/4 для кластера из четырёх узлов
• 3/5 для кластера из пяти узлов

Выбор узлов-реплик (множества from) для каждого <ID> является детерминированным:

• Множество узлов-реплик для фиксированного ID и меньшего значения from является строгим подмножеством для большего значения from
• При PUT не сохраняется больше копий данных, чем указано в from (т.е. не стоит писать лишние копии данных на все реплики)

Фактически, с помощью параметра replicas клиент выбирает, сколько копий данных он хочет хранить, а также уровень консистентности при выполнении последовательности операций для одного ID.

Таким образом, обеспечиваются следующие примеры инвариантов (список не исчерпывающий):

• GET с 1/2 всегда вернёт данные, сохранённые с помощью PUT с 2/2 (даже при недоступности одной реплики при GET)
• GET с 2/3 всегда вернёт данные, сохранённые с помощью PUT с 2/3 (даже при недоступности одной реплики при GET)
• GET с 1/2 "увидит" результат DELETE с 2/2 (даже при недоступности одной реплики при GET)
• GET с 2/3 "увидит" результат DELETE с 2/3 (даже при недоступности одной реплики при GET)
• GET с 1/2 может не "увидеть" результат PUT с 1/2
• GET с 1/3 может не "увидеть" результат PUT с 2/3
• GET с 1/2 может вернуть данные несмотря на предшествующий DELETE с 1/2
• GET с 1/3 может вернуть данные несмотря на предшествующий DELETE с 2/3
• GET с ack равным quorum(from) "увидит" результат PUT/DELETE с ack равным quorum(from) даже при недоступности < quorum(from) реплик

Report

Присылайте pull request со своей реализацией поддержки кластерной конфигурации на review. Не забудьте нагрузить, отпрофилировать и проанализировать результаты профилирования под нагрузкой. С учётом репликации набор тестов расширяется, поэтому не забывайте подмёрдживать upstream.

Этап 6. Асинхронный клиент (deadline 2020-11-04 00:00:00 MSK)

Переключаем внутреннее взаимодействие узлов на асинхронный java.net.http.HttpClient. Параллельно отправляем запросы репликам и собираем подтверждения на CompletableFuture.

Проведите нагрузочное тестирование с помощью wrk в несколько соединений.

Отпрофилируйте приложение (CPU, alloc и lock) под нагрузкой с помощью async-profiler и сравните результаты latency и профилирования по сравнению с неасинхронной версией.

Report

Присылайте pull request со своей реализацией на review, а также результатами нагрузочного тестирования через wrk2, профилирования под нагрузкой и качественным анализом по сравнению с предыдущей синхронной реализацией.

Этап 7. Range-запросы (deadline 2020-11-18 00:00:00 MSK)

Реализуйте получение диапазона данных с помощью HTTP GET /v0/entities?start=<ID>[&end=<ID>], который возвращает:

• Статус код 200 OK
• Возможно пустой отсортированный (по ключу) набор ключей и значений в диапазоне ключей от обязательного start (включая) до опционального end (не включая)
• Используйте Chunked transfer encoding
• Чанки в формате <key>\n<value>

Диапазон должен отдаваться в потоковом режиме без формирования всего ответа в памяти.

Report

После прохождения модульных тестов, присылайте pull request с изменениями.

Этап 8. Бонусный (deadline 2020-11-25 00:00:00 MSK)

Фичи, которые позволяют получить дополнительные баллы (при условии добавления набора тестов, демонстрирующих корректность, где применимо):

• Развёрнутая конструктивная обратная связь по курсу: достоинства и недостатки курса, сложность тем, предложения по улучшению
• Кластерные range-запросы с учётом шардирования и репликации
• Read repair при обнаружении расхождений между репликами
• Expire: возможность указания времени жизни записей
• Server-side processing: трансформация данных с помощью скрипта, запускаемого на узлах кластера через API
• Нагрузочное тестирование при помощи Y!CSB
• Нагрузочное тестирование при помощи Yandex.Tank
• Регулярный автоматический фоновый compaction (модульные и нагрузочные тесты)
• Hinted handoff по аналогии с Cassandra
• Устранение неконсистентностей между репликами по аналогии с Cassandra nodetool repair, например, на основе Merkle Tree
• Тесты для consistent/rendezvous hashing, демонстрирующие равномерность распределения данных
• Блочная компрессия данных на основе LZ4
• Что-нибудь своё?

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