Cassandra

Всем привет. Уже сейчас в OTUS открывает набор в новую группу курса Highload Architect. В связи с этим я продолжаю серию своих публикаций, написанных специально для этого курса, а также приглашаю вас на свой бесплатный демо урок по теме: Индексы в MySQL: best practices и подводные камни. Записаться на вебинар можно тут.

---

Введение

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

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

Паттерн Сага

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

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

Типов транзакций в саге несколько, целых четыре:

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

Организовывать сагу можно с помощью хореографии или оркестрации.

В случае с хореографической саги выделенный оркестратор отсутствует. На примере сервиса заказов и пользователей она может выглядеть так: сервис заказов получает запрос и создает заказ в состоянии PENDING, а затем публикует событие Заказ создан. Обработчик событий в сервисе пользователей обрабатывает данное событие, пытается зарезервировать товар и публикует результат в виде события. Сервис заказов обрабывает данное событие, подтверждая или отменяя заказ в зависимости от прочитанного результата.

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

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

Сага позволяет добиться ACD-модели (Atomicity + Consistency + Durability в терминах ACID), но одну букву мы потеряли. Недостаток буквы I приводит к известным проблемам недостатка изолированности. К ним относятся: потерянные обновления (lost updates) одна сага перезаписывает изменения, внесенные другой, не читая их при этом, грязное чтение (dirty reads) транзакция или сага читают незавершенные обновления другой саги, нечеткое/неповторяемое чтение (fuzzy/nonrepeatable reads) два разных этапа саги читают одни и те же данные, но получают разные результаты, потому что другая сага внесла изменения. Существует ряд паттернов, позволяющих пофиксить те или иные аномалии: семантическая блокировка, коммутативные обновления, пессимистическое представление, повторное чтение значения, файл изменений и по значению. Вопрос обеспечения изоляции остается открытым.

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

Заключение

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

Всем привет. Уже в сентябре OTUS открывает набор в новую группу курса Highload Architect. В связи с этим я продолжаю серию своих публикаций, написанных специально для этого курса, а также приглашаю вас на свой бесплатный вебинар, в рамках которого я подробно расскажу о программе курса и формате обучения в OTUS. Записаться на вебинар можно тут.

---

Введение

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

Согласованность

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

Причина проблемы

Почему обеспечение согласованности затруднено именно в микросервисной архитектуре? Дело в том, что данный архитектурный стиль зачастую предполагает применение паттерна database per service. Позволю себе напомнить, что этот паттерн заключается в том, что у каждого микросервиса своя независимая база или базы (базы, потому что помимо первичного источника данных может использоваться, например, кеш). Такой подход позволяет с одной стороны не добавлять неявные связи по формату данных между микросервисами (микросервисы взаимодействуют только явно через API), с другой стороны по максимуму использовать такое преимущество микросервисной архитектуры как technology agnostic (мы можем выбирать подходящую под особую нагрузку на микросервис технологию хранения данных). Но при всем при этом мы потеряли гарантию согласованности данных. Посудите сами, монолит общался с одной большой базой, которая предоставляла возможности по обеспечению ACID транзакций. Теперь баз данных стало много, а вместо одной большой ACID транзакции у нас много небольших ACID транзакций. Нашей задачей будет объединение всех этих транзакций в одну распределенную.

Оптимистичная согласованность

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

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

Варианты обеспечения консистентности

Если для бизнеса все-таки согласованность является критичной, можно попытаться ее обеспечить несколькими способами. Если мы говорим о ситуации, когда данные обновляются одним сервисом (например имеет место репликация базы данных), то можно применить стандартные алгоритмы обеспечения консистентности такие как Paxos или Raft. Такие транзакции называются гомогенными. Если данные обновляются несколькими сервисами (то есть имеет место гетерогенная транзакция), то как тут как раз и начинаются сложности, о которых мы говорили выше.

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

Двухфазный коммит

Механизм предельно прост: есть некоторый transaction manager, который собственно оркестрирует транзакцию. На первом этапе (prepare) transaction manager подает соответствующую команду для resource manager'ов, по которой они в свои журналы записывают данные, которые будут закоммичены. Получив подтверждение ото всех resource manager'ов об успешном завершении первого этапа, transaction manager начинает второй этап и подает следующую команду (commit), по которой resource manager'ы применяют принятые ранее изменения.

Несмотря на кажущуюся простоту, такой подход обладает рядом недостатков. Во-первых, если хотя бы один resource manager даст сбой на втором этапе, вся транзакция должна быть отменена. Таким образом, нарушается один из принципов микросервисной архитектуры устойчивость к отказам (когда мы приходили к распределенной системе, мы сразу закладывались на то, что отказ в ней является нормой а не исключительной ситуацией). Более того, если отказов будет много (а их будет много), то процесс отмены транзакций необходимо будет автоматизировать (в том числе и писать транзакции, откатывающие транзакции). Во-вторых, сам transaction manager является единой точкой отказа. Он должен уметь транзакционно выдавать id-шники транзакциям. В-третьих, поскольку хранилищу подаются специальные команды, логично предположить, что хранилище должно уметь это делать, то есть соответствовать стандарту XA, а не все современные технологии ему соответствуют (такие брокеры как Kafka, RabbitMQ и NoSQL решения как MongoDB и Cassandra не поддерживают двухфазные коммиты).

Вывод, напрашивающийся из всех этих факторов, был отлично сформулирован Крисом Ричардсоном: 2PC not an option (двухфазный коммит не вариант).

Вывод

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

---

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

---

Читать ещё:

• Сортировка выбором
• Блокировки как один из способов обеспечения изоляции транзакций
• MVCC как один из способов обеспечения изоляции транзакций
• Что нужно знать об устройстве коллекций, основанных на хешировании
  
• Почему может понадобиться полусинхронная репликация?
• Сортировка вставками

Apache Cassandra 4.0 приближается к бете (прим. переводчика: на текущий момент уже доступна бета 4, выпущенная в конце декабря 2020), и это первая версия, которая будет поддерживать JDK 11 и более поздних версий. Пользователей Apache Cassandra, очевидно, волнует задержка, так что мы возлагаем большие надежды на ZGC новый сборщик мусора с низкой задержкой, представленный в JDK 11.

В JDK 14 он был выпущен уже в GA-версии, и нам было очень интересно оценить, насколько он подходит для кластеров Apache Cassandra. Мы хотели сравнить производительность Apache Cassandra 3.11.6 и 4.0 и проверить, подходит ли Shenandoah, сборщик мусора от Red Hat, для продакшена. Спойлер: Cassandra 4.0 значительно лучше по производительности сама по себе, а с новыми сборщиками мусора (ZGC и особенно Shenandoah) будет совсем хорошо.

Методология бенчмарков

Мы провели следующие бенчмарки, используя утилиту tlp-cluster для развертывания и настройки кластеров Apache Cassandra в AWS и tlp-stress для создания нагрузок и сбора метрик. Все эти инструменты доступны в опенсорсе, все бенчмарки легко воспроизвести при наличии учетки AWS.

Кластеры состояли из трех нод на инстансах r3.2xlarge и одной стресс-ноды на c3.2xlarge.

Мы использовали дефолтные параметры Apache Cassandra, кроме настроек кучи и сборщика мусора.

Для развертывания и настройки кластера мы взяли последний релиз tlp-cluster. Недавно мы добавили скрипты, чтобы автоматизировать создание кластера и установку Reaper и Medusa.

Установите и настройте tlp-cluster по инструкциям в документации, чтобы создать такие же кластеры, которые мы использовали для бенчмарков:

# 3.11.6 CMS JDK8build_cluster.sh -n CMS_3-11-6_jdk8 -v 3.11.6 --heap=16 --gc=CMS -s 1 -i r3.2xlarge --jdk=8 --cores=8# 3.11.6 G1 JDK8build_cluster.sh -n G1_3-11-6_jdk8 -v 3.11.6 --heap=31 --gc=G1 -s 1 -i r3.2xlarge --jdk=8 --cores=8# 4.0 CMS JDK11build_cluster.sh -n CMS_4-0_jdk11 -v 4.0~alpha4 --heap=16 --gc=CMS -s 1 -i r3.2xlarge --jdk=11 --cores=8# 4.0 G1 JDK14build_cluster.sh -n G1_4-0_jdk14 -v 4.0~alpha4 --heap=31 --gc=G1 -s 1 -i r3.2xlarge --jdk=14 --cores=8# 4.0 ZGC JDK11build_cluster.sh -n ZGC_4-0_jdk11 -v 4.0~alpha4 --heap=31 --gc=ZGC -s 1 -i r3.2xlarge --jdk=11 --cores=8# 4.0 ZGC JDK14build_cluster.sh -n ZGC_4-0_jdk14 -v 4.0~alpha4 --heap=31 --gc=ZGC -s 1 -i r3.2xlarge --jdk=14 --cores=8# 4.0 Shenandoah JDK11build_cluster.sh -n Shenandoah_4-0_jdk11 -v 4.0~alpha4 --heap=31 --gc=Shenandoah -s 1 -i r3.2xlarge --jdk=11 --cores=8

Примечание. Чтобы воспроизвести схожие условия, мы использовали для всех тестов один набор инстансов EC2.

Мы сделали апгрейд с Cassandra 3.11.6 на Cassandra 4.0~alpha4 и меняли JDK следующим кодом:

#!/usr/bin/env bashOLD=$1NEW=$2curl -sL https://github.com/shyiko/jabba/raw/master/install.sh | bash. ~/.jabba/jabba.shjabba uninstall $OLDjabba install $NEWjabba alias default $NEWsudo update-alternatives --install /usr/bin/java java ${JAVA_HOME%*/}/bin/java 20000sudo update-alternatives --install /usr/bin/javac javac ${JAVA_HOME%*/}/bin/javac 20000

Мы использовали следующие значения JDK при вызове jabba:

• openjdk@1.11.0-2
• openjdk@1.14.0
• openjdk-shenandoah@1.8.0
• openjdk-shenandoah@1.11.0

OpenJDK 8 был установлен с помощью Ubuntu apt.

Вывод команды java -version для разных JDK, которые мы использовали для бенчмарков:

jdk8

openjdk version "1.8.0_252"OpenJDK Runtime Environment (build 1.8.0_252-8u252-b09-1~18.04-b09)OpenJDK 64-Bit Server VM (build 25.252-b09, mixed mode)

jdk8 с Shenandoah

openjdk version "1.8.0-builds.shipilev.net-openjdk-shenandoah-jdk8-b712-20200629"OpenJDK Runtime Environment (build 1.8.0-builds.shipilev.net-openjdk-shenandoah-jdk8-b712-20200629-b712)OpenJDK 64-Bit Server VM (build 25.71-b712, mixed mode)

jdk11

openjdk version "11.0.2" 2019-01-15OpenJDK Runtime Environment 18.9 (build 11.0.2+9)OpenJDK 64-Bit Server VM 18.9 (build 11.0.2+9, mixed mode)

jdk11 с Shenandoah

openjdk version "11.0.8-testing" 2020-07-14OpenJDK Runtime Environment (build 11.0.8-testing+0-builds.shipilev.net-openjdk-shenandoah-jdk11-b277-20200624)OpenJDK 64-Bit Server VM (build 11.0.8-testing+0-builds.shipilev.net-openjdk-shenandoah-jdk11-b277-20200624, mixed mode)

jdk14

openjdk version "14.0.1" 2020-04-14OpenJDK Runtime Environment (build 14.0.1+7)OpenJDK 64-Bit Server VM (build 14.0.1+7, mixed mode, sharing)

CMS

CMS (Concurrent Mark Sweep) это текущий дефолтный сборщик мусора в Apache Cassandra. Он был удален из JDK 14, так что все тесты проводились с JDK 8 или 11.

Для CMS параметры были такие:

-XX:+UseParNewGC-XX:+UseConcMarkSweepGC-XX:+CMSParallelRemarkEnabled-XX:SurvivorRatio=8-XX:MaxTenuringThreshold=1-XX:CMSInitiatingOccupancyFraction=75-XX:+UseCMSInitiatingOccupancyOnly-XX:CMSWaitDuration=10000-XX:+CMSParallelInitialMarkEnabled-XX:+CMSEdenChunksRecordAlways-XX:+CMSClassUnloadingEnabled-XX:ParallelGCThreads=8-XX:ConcGCThreads=8-Xms16G-Xmx16G-Xmn8G

Флаг -XX:+UseParNewGC удален из JDK 11 и является неявным. С этим флагом JVM не запустится.

Для CMS мы задали максимальный размер кучи 16 ГБ, иначе между сборками будут слишком длинные паузы.

G1

G1GC (сборщик мусора по принципу Garbage-First) настроить легче, чем CMS, потому что он динамически изменяет размер младшего поколения, но лучше работает с большими кучами (от 24 ГБ). Это объясняет, почему он не стал дефолтным сборщиком мусора. Задержки у него выше, чем у настроенного CMS, зато пропускная способность лучше.

Для G1 параметры были такие:

-XX:+UseG1GC-XX:G1RSetUpdatingPauseTimePercent=5-XX:MaxGCPauseMillis=300-XX:InitiatingHeapOccupancyPercent=70-XX:ParallelGCThreads=8-XX:ConcGCThreads=8-Xms31G-Xmx31G

Для версии 4.0 мы использовали JDK 14 в тестах с G1.

Мы взяли размер кучи 31 ГБ, чтобы использовать преимущества compressed oops и получить больше адресуемых объектов для кучи минимального размера.

ZGC

ZGC (Z Garbage Collector) это новейший сборщик мусора в JDK, который минимизирует задержку благодаря паузам stop-the-world меньше 10 мс. Предполагается, что благодаря этому размер кучи не будет влиять на продолжительность пауз, что позволит увеличить его до 16 ТБ. Если так и будет, хранилище вне кучи не понадобится, и некоторые аспекты разработки в Apache Cassandra станут проще.

Для ZGC параметры были такие:

-XX:+UnlockExperimentalVMOptions-XX:+UseZGC-XX:ConcGCThreads=8-XX:ParallelGCThreads=8-XX:+UseTransparentHugePages-verbose:gc-Xms31G-Xmx31G

-XX:+UseTransparentHugePages здесь нужен как обходной путь, чтобы избежать включения больших страниц в Linux. В официальной документации по ZGC сказано, что потенциально могут возникать пики задержки, но на практике мы такого не увидели. Может быть, следует протестировать пропускную способность на больших страницах и посмотреть, как это скажется на результате.

ZGC не может использовать compressed oops и не соблюдает порог 32 ГБ. Мы взяли кучу размером 31 ГБ, как и для G1, чтобы у системы был тот же объем свободной оперативки.

Shenandoah

Shenandoah это сборщик мусора с низкой задержкой от Red Hat. Он доступен как бэкпорт в JDK 8 и 11 и в mainline-сборках OpenJDK начиная с Java 13.

Как и ZGC, Shenandoah стремится к параллелизму, чтобы паузы не зависели от размера кучи.
Для Shenandoah параметры были такие:

-XX:+UnlockExperimentalVMOptions-XX:+UseShenandoahGC-XX:ConcGCThreads=8-XX:ParallelGCThreads=8-XX:+UseTransparentHugePages-Xms31G-Xmx31G

Shenandoah может использовать compressed oops, а значит лучше работает с кучами размером чуть меньше 32 ГБ.

Конфигурация Cassandra 4.0 JVM

Cassandra 4.0 поставляется с отдельными файлами jvm.options для Java 8 и Java 11, а именно:

• conf/jvm-server.options
• conf/jvm8-server.options
• conf/jvm11-server.options

Апгрейд до 4.0 будет работать с имеющимся файлом jvm.options версии 3.11, если его переименовать в jvm-server.options, а файлы jvm8-server.options и jvm11-server.options удалить. Но мы такой подход не рекомендуем.

Лучше повторно применить параметры из предыдущего файла jvm.options к новым файлам jvm-server.options и jvm8-server.options. Конкретные файлы опций Java, в основном, связаны с флагами для сборки мусора. Если обновить эти два файла, будет проще настроить jvm11-server.options и перейти с JDK 8 на JDK 11.

Рабочие нагрузки

В бенчмарках мы использовали восемь потоков с ограничением скорости и соотношением записи и чтения 80/20%. tlp-stress широко использует асинхронные запросы, которые легко могут перегрузить ноды Cassandra с ограниченным числом стресс-потоков. В нагрузочных тестах каждый поток отправлял 50 параллельных запросов за раз. Мы создали keyspace с коэффициентом репликации 3, и все запросы выполнялись на уровне согласованности LOCAL_ONE.

Все сборщики мусора и версии Cassandra тестировались по нарастанию 25к, 40к, 45к и 50к операций в секунду, чтобы оценить производительность при разных уровнях нагрузки.

Команды tlp-stress:

tlp-stress run BasicTimeSeries -d 30m -p 100M -c 50 --pg sequence -t 8 -r 0.2 --rate <desired rate> --populate 200000

Все рабочие нагрузки выполнялись 30 минут, загружая от 5 до 16 ГБ данных на ноду и допуская нагрузку от compaction в разумной степени.

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

Результаты бенчмарков

3.11.6, 25-40к операций/с:

4.0, 25-40к операций/с:

4.0, 45-50к операций/с:

В плане пропускной способности Cassandra 3.11.6 осилила 41к операций в секунду, а Cassandra 4.0 51к, то есть на 25% больше. В обоих случаях использовался CMS. В версии 4.0 много улучшений, которые объясняют эти результаты, особенно в вопросах нагрузки на кучу, вызванной compaction (см. CASSANDRA-14654, например).

Shenandoah в jdk8 на Cassandra 3.11.6 не достигает максимальной пропускной способности в тесте на 40к операций в секунду появляются невыполненные запросы. С jdk11 и Cassandra 4.0 результат получше 49,6к, почти как у CMS. У G1 и Shenandoah с jdk 8 получилось 36к/с на Cassandra 3.11.6.

G1, видимо, усовершенствован в jdk14 и показал себя немного лучше, чем с jdk11, 47 против 50к/с.

ZGC не дотянул до конкурентов ни с jdk11, ни с jdk14, показав 41к/с максимум.

При умеренной нагрузке Shenandoah в jdk8 показывает хороший результат на Cassandra 3.11.6, но под высокой нагрузкой серьезно увеличиваются задержки.

С CMS Cassandra 4.0 показывает среднюю задержку для p99 (99-го перцентиля) от 11 до 31 мс при 50к операций в секунду. Средняя задержка операций чтения для p99 при умеренной нагрузке снизилась с 17 мс в Cassandra 3.11.6 до 11,5 мс в Cassandra 4.0, то есть на целых 30%.

В целом Cassandra 4.0 лучше Cassandra 3.11.6 с теми же сборщиками мусора на 2530% по пропускной способности и задержке.

Shenandoah дает очень низкие задержки при умеренной нагрузке в Cassandra 3.11.6, но с высокими нагрузками справляется не очень.

По задержкам ZGC хорошо себя проявляет при умеренной нагрузке, особенно с jdk14, но при более высокой скорости отстает от Shenandoah. Почти во всех нагрузочных тестах Shenandoah показал минимальные средние задержки для p99 для чтения и записи. Если сложить эти задержки с пропускной способностью в Cassandra 4.0, получится что этот сборщик мусора неплохой выбор, если вы обновляете версию. Средняя задержка 2,64 мс при чтении для p99 при умеренной нагрузке это впечатляет. Тем более на клиенте.

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

При умеренной нагрузке Shenandoah дает среднюю задержку для p99 на 77% ниже. 2,64 мс это самый низкий результат. Полезно для сценариев, где важна задержка. По сравнению с CMS в Cassandra 3.11.6 это на целых 85% ниже для операций чтения на p99!

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

Итоги

G1 улучшает юзабилити Cassandra не нужно настраивать размеры поколений в ущерб производительности. В версии Apache Cassandra 4.0, которая и без того работает гораздо лучше, можно будет использовать сборщики мусора нового поколения, например Shenandoah и ZGC, которые не требуют тщательной настройки и сокращают задержки.

Мы не рекомендуем Shenandoah для Cassandra 3.11.6, поскольку он не справляется с высокими нагрузками, но начиная с jdk11 и Cassandra 4.0 этот сборщик мусора показывает низкие задержки и обеспечивает почти максимальную пропускную способность для базы данных.

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

Загрузите последнюю сборку Apache 4 и попробуйте сами. Будем рады обратной связи тут или в Slack компании ASF.

От редакции: приглашаем на открытую онлайн-конференцию Cassandra Day Russia 2021 в субботу 27 марта. В программе доклады и воркшопы от опытных NoSQL-специалистов.

Yelp это крупнейшее в США приложение для заказа еды и услуг. Оно установлено более чем на 30 млн уникальных устройств, в нём зарегистрировано более 5 млн. компаний. Для хранения и доступа к данным в Yelp используют Cassandra. Как и для каких задач применяется эта база данных, на конференции Cassandra Day Russia 2021 рассказал Александр Широков, Database Reliability Engineer в Yelp.

Cassandra это база данных из семейства NoSQL. Согласно полному определению, это distributed wide-column NoSQL datastore. Cassandra оптимизирована для writeов, и consistency запросов можно двигать на очень низком уровне.

Не углубляясь в детали, я расскажу, как мы используем Cassandra в Yelp. У нас это один из самых популярных инструментов. Мы применяем MySQL и Cassandra. Если MySQL все девелоперы знают, то Cassandra нет. Но это то, что мы используем постоянно и на довольно большом масштабе.

Я расскажу:

1. Как мы автоматизируем изменение схемы: как девелоперы изменяют таблицы, добавляют таблицы или колонки. Если позволить им делать это по своему усмотрению, то получается очень много проблем. Чтобы их избежать, мы этот процесс автоматизировали.
2. Как разработчик получает доступ к Cassandra. Один из вариантов находить каждый инстанс Cassandra-кластера через host и port, но это решение не масштабируется. Я покажу статистику по нашим кластерам и вы увидите, что на нашем масштабе нужно что-то другое.
3. Закончу одним из проектов, над которым мы работаем сейчас, и который выглядит довольно многообещающим.

Как используется Cassandra в Yelp

В любой месяц отправляется около 90 млрд запросов, в любой момент времени задеплоено около 60 Kubernetes-кластеров и около 500 подов (один под один инстанс Cassandra).

С точки зрения нагрузки на кластеры, наша самая высокая Read Load около 600 тыс. RPS на один из наших кластеров, Write Load около 400 RPS. В целом довольно высокая нагрузка.

В Yelp мы поддерживаем широкий спектр пользовательских кейсов. Мы поддерживаем и аналитический процессинг, и процессинг транзакций. Например, есть какой-то batch-сервис, онлайн или веб-сервис, который отправляет запросы в Cassandra и что-то пишет в Cassandra.

Также наша стриминг-инфраструктура зависит от Cassandra. Плюс мы собираем в Cassandra данные для аналитики, агрегируем какие-то метрики; от Cassandra зависит наша caching-инфраструктура и distributed tracing возможность для разработчиков трейсить жизненный цикл реквеста в Yelp, смотреть, как реквест движется между микросервисами (трейсинг инфраструктуры).

Как вы видите, спектр юзкейсов довольно широкий.

Мы деплоим Cassandra в основном в двух регионах: us-east и us-west.

Несмотря на то, что мы деплоим на us-east и us-west, каждый наш кластер называется multi region cluster. Если клиент хочет написать в us-east, а тот недоступен, тогда он напишет в us-west.

В каждом регионе мы деплоим (так как у нас инфраструктура основана на Kubernetes) наш собственный разработанный Kubernetes-оператор. Роль этого оператора мы даём ему кастомные ресурсы смотреть на Cassandra: количество подов, которые должны быть в любой момент времени, или какие-то значения CPU и памяти. Оператор просто сравнивает те значения, которые мы ему даём, с тем, что происходит на самом деле мониторит кластер. И если случается несовпадение, то оператор поднимет ещё один под.

Например, оператору сказали, что нужно пять подов, а сейчас задеплоено только четыре, он поднимет ещё один. Это снимает много работы с нас, как с команды, которая занимается поддержкой инфраструктуры. Синхронизация происходит с помощью etcd. Etcd это простенький distributed key-value store.

Мы деплоим всё на AWS, для хранения самих данных используем EBS volumes Amazon Elastic Block Storage. Вся инфраструктура основана на Kubernetes, плюс мы разработали собственный оператор.

Для подключения клиента к каждому кластеру, каждому инстансу кластера мы используем продукт Smartstack, а в следующей итерации Envoy.

Как в Yelp делают изменение схемы (Schema Changes)

С чего можно начать автоматизировать этот процесс: можно взять все схемы (schema), которые есть, и засунуть в один репозиторий самый простой вариант. И потом на каждой Cassandra node запустить процесс, который будет смотреть на то, что у нас в репозитории и то, что у нас в Cassandra, сравнивать, и если у нас тут есть какая-нибудь схема (например, какая-то таблица), а в кластере Cassandra она не существует, она просто создаст новую таблицу так, чтобы они просто совпадали.
Если, например, несовпадение, так, чтобы мы знали, что происходит, что делает этот процесс, процесс файлит Jira-тикет.

Это то как мы начинали. Но в этом есть несколько проблем: довольно сложно поддерживать Alter Table стейтменты. Сложно понять, куда это вообще, как это поддерживать. Разработчики не знали про это. Каждый раз, когда разработчикам нужно было изменить таблицу, они шли к нам и спрашивали. Это такая проблема, о которой нам нужно было бы рассказывать разработчикам каждый раз. Можно было бы, конечно, записать это куда-то в документацию, но у кого есть время её читать?

Если разработчики создают какую-то схему, они засабмитят в гит-репозиторий, но у них никакого фидбека, нет обратной связи, когда эта схема будет применена на сам кластер, потому что процесс, который запускается на Cassandra, он, естественно, не постоянный, как butch-процесс, который периодически пулит и пытается применить изменения. Также изменения схем были применены в очень рандомном порядке в наших окружениях. То есть у нас есть несколько окружений: development, staging, production, и возможен был такой вариант, когда production была применена ранее, чем development.

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

Если нам нужно было ревью каждого изменения схемы, не было процессов, когда мы, как инженеры по Database Reliability, обязательно бы это делали. И в том числе, когда разработчики пишут их SQL-команды, чтобы создать таблицу, например, не было никакой обратной связи с точки зрения ошибок: они написали, но не было возможности им сказать, что, например, вот этот тип колонки не поддерживается, или вы что-то делаете не так.

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

Как всё работает сейчас

Для начала надо дать терминологию. В Yelp используется pushplan это файл, который хранит инструкции по тому, как задеплоить какой-то код. Мы используем это в довольно широком контексте. Например, в контексте любой инфраструктурной работы. Если нам надо изменить часть инфраструктуры, добавить значение в конфиг, то мы обычно пишем pushplan. И эта pushplan-терминология используется и в случае с Cassandra.

Расскажу буквально в паре слов, как процесс выглядит сейчас с точки зрения инженера в Yelp. Если они хотят создать новую таблицу или новый keyspace в Cassandra, они создают папку pushplan и на каждом dev-боксе, который инженер использует, есть utility-функция или команда, которая называется pushplan. Они создают этот pushplan, этот текстовый файл с инструкциями, где они задают буквально пару значений: Jira-тикет, кластер, где они хотят создать, название keyspace, название таблицы и тип базы данных. В итоге эта функция генерирует SQL-файл.

Затем они пишут саму SQL-команду. То есть разработчики пишут, например, create table. Когда они пытаются сгенерировать pushplan, мы тут же можем применить так как всё автоматизировано, у нас есть контроль по тому, что мы можем делать с SQL-командами.

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

Разработчик исправляет ошибку, генерирует файл снова, теперь всё сработало отлично.

Пример файла, который сгенерируется: есть уникальный ID, имя автора, время, версия и сама команда.

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

После того, как мы замёрджили коммит, девелопер получает уведомление в слаке с результатами его pushplan. То есть он может увидеть в реальном времени, что их команда применилась сначала на окружение dev, потом на stage и prod. И видит, когда это случилось. Разработчики также могут запрашивать статус и видеть, на каких окружениях уже применяется и на каких нет. Если где-то ошибка, то могут увидеть, где она произошла. Для каждого кластера можно видеть историю, как изменялась схема всех таблиц.

Какие преимущества мы получили:

• Разработчикам не надо вручную тестировать SQL-команды. Им не нужно переживать, что они не смогут исправить ошибку, если вдруг допустят её. Мы дадим им фидбек сразу, как только они напишут, довольно автоматизированным способом.
• Мы как DRE-команда можем делиться best practices. Мы можем им сказать, например, что каждая таблица должна содержать как минимум один primary key; не нужно изменять какие-то колонки или что 20 колонок в одном pushplan это не лучшее решение. Мы также можем добавить правила проверки.

Вот так это и работает.

Cassandra для разработчиков

Теперь я немного расскажу, как Cassandra используется разработчиками: как они получают доступ к Cassandra-кластерам или Cassandra-инстансам.

Снова начну с истории. Около семи или восьми лет назад, когда мы начали использовать Cassandra, у нас был один главный web application и буквально несколько сервисов.

Потом мы задеплоили несколько Cassandra-кластеров и каждый сервис или каждая команда, которая работала с этим сервисом, должна была искать Cassandra-кластер вручную: искать host-порт, забивать это вручную в конфиг. С другой стороны, им тоже нужно было искать какой-то драйвер, чтобы отправлять запросы на Cassandra, эти драйверы имели разную версию получалась очень беспорядочная ситуация. И как вы видите, нагрузка между кластерами распределялась неравномерно.

К чему мы в итоге пришли довольно рано, это мы задеплоили proxy-сервис, который мы назвали Apollo. По сути это очень простой сервис на Python, который находится между сервисами и между Cassandra-кластерами.

Эта абстракция позволила нам решить несколько интересных проблем. Во-первых, разработчикам больше не нужно было думать, как подключиться к Cassandra, какие host и port, не искать всё это вручную, не поддерживать драйверы Cassandra. То есть вся сложность инфраструктуры была абстрагирована от разработчиков. Всё что им нужно знать это host и port сервиса Apollo.

Мы используем SmartStack, который я уже упоминал ранее, чтобы автоматически найти инстанс Apollo. То есть сервисам даже не нужно искать host и port, им просто нужно сказать: подключись к Apollo, и SmartStack найдёт нужный инстанс.

Это одна проблема, которую мы решили.

Другая решённая проблема это то, что сейчас мы, как команда DRE, могли производить эксперименты над нашим Cassandra-кластером. Мы могли, например, изменить версию драйвера Cassandra, если нам хотелось, или сделать scale up или scale down кластера, в зависимости от нагрузки, которую мы получаем. Это позволило нам равномерно распределить нагрузку на кластеры.

То есть одним proxy-сервисом мы решили несколько проблем.

Также мы могли изолировать клиентов в их собственный Apollo-кластер. Если, например, есть какие-то клиенты или сервисы, юзкейсы которых очень важны для бизнеса, мы просто изолируем их в отдельный кластер.

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

Идея сервиса в том, чтобы разработчики могли там создавать свой код добавлять бизнес-логику И мы заметили буквально недавно, что сервис вырос и сейчас поддерживает где-то около 140 юзкейсов и коннектится практически к 40 кластерам. И этот сервис хранит около 60 тыс. строк только бизнес-логики. И каждый раз, когда мы деплоим, мы раним 60 тысяч строк integration-тестов, что занимает около часа. Масштаб сервиса расширился до сотен инстансов.

Если разработчикам нужно написать самую простую команду, select-команду, им нужно написать довольно много кода.

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

Так же нужно писать unit-тесты, нужно писать swagger spec, потому что каждый клиент это end point или API в Apollo, и надо писать fake data и конфиги чтобы end-to-end тесты работали правильно.

Мы заметили, что (как и с pushplan) то, что работало вначале, сейчас не работает. Просто потому, что масштаб уже совсем другой. Мы заметили, что нужен какой-то другой, новый подход к решению этой проблемы, но мы не хотели начинать с абсолютного нуля. У Apollo есть некоторые преимущества, которые решали интересные проблемы: например, изоляция сложности в доступе и управлении инфраструктурой. Это важная проблема.

Но с другой стороны, с каким-то новым продуктом, с заменой Apollo, решить несколько проблем:

1. Увеличить скорость итераций над клиентским кодом. Избавить инженеров от необходимости писать так много кода.
2. Облегчить доступ к Cassandra-кластеру, ну и в целом модернизировать технологии, которые существовали.

В итоге мы пришли к продукту, который называется Stargate. Это Open Source продукт, который сделан и используется компанией DataStax. Про него можно думать как про API для базы данных. По сути это очень лёгкий слой между слоем приложения и базой данных.

Как это работает

Вы можете думать о Cassandra-кластере, как о коллекции нод, инстансов, организованных в кольцо. Когда приходит запрос от клиента, одна из нод выбирается как координатор, и её задача скоординировать запрос. Выбрать метрики, которые приходят от consistency requirements, от клиента, и отправить результат обратно клиенту.

Как это работает со Stargate. Stargate присоединяется к Cassandra-кластеру как координатор нод. И его задача просто скоординировать запрос: найти реплики Но единственное отличие в том, что Stargate не хранит никакие данные. То есть он решает две проблемы: работает как координатор, и он очень лёгкий просто потому, что ему не нужно думать о том, как хранить данные. Мы добавили небольшой proxy поверх Stargate.

Ну и вообще, Stargate можно думать об этом слое поверх Cassandra, который работает как координатор, но также предоставляет API для разработчиков. Он предоставляет REST API, можно коннектиться через gRPC, WebSocket или GraphQL.

Мы выбрали GraphQL API.

Почему GraphQL

Stargate берёт Cassandra schema все таблицы, которые в key space существуют, и автоматически генерирует GraphQL-запросы. Что это даёт разработчикам: Type-safe API. GraphQL решает очень интересную проблему over и under fetching. То есть клиенты могут очень чётко сказать: нам нужно вот такие колонки возвратить. Например, то что раньше было сложнее сделать с Apollo. И также мы можем запускать несколько Cassandra-запросов в одном запросе. Если нам нужно, например, синхронизировать какие-то таблицы в разных кластерах в одном запросе.

В целом GraphQL становится новым стандартом в API development и используется в больших технологических компаниях. Например, в Facebook (откуда он и произошёл), в Twitter, Github. В Yelp мы тоже довольно давно используем GraphQL.

Что дальше

Расскажу, что мы планируем делать с проектом дальше. Надеюсь, это вам даст немного больше информации о том, как мы делаем какие-то вещи в Yelp. Расскажу, как мы задеплоили MVP (взяли Stargate и просто задеплоили одну ноду) и потом расскажу немного про клиентские библиотеки то, как микросервисы взаимодействуют друг с другом в Yelp.

Тестирование Stargate

Перед тем как начать любую работу по замене Apollo, мы взяли Stargate и подумали: а что если мы задеплоим одну ноду (один node), которая присоединится к Cassandra-кластеру из трёх реплик, и потом запустим тестирование нагрузки запустим 100 тредов, поднимем нагрузку до 3 тыс. RPS и будем ранить это в течение 24 часов. Что тогда произойдёт? А что если потестим парочку сценариев: зарестартим один из Cassandra нодов или зарестартим сам инстанс? Что произойдёт тогда?

На графике видно, что мы проводили этот тест в течение 24 часов и подняли нагрузку примерно до 3 тыс. RPS.

Если посмотреть на rage of latencies всех запросов, то они довольно в допустимом пределе. Если посмотреть на высокий хвост (high tail) задержек (99.9+), то он немного spite.

Это дало нам несколько интересных инсайтов по тому, как Stargate работает:

1. Да, память стабильная, CPU повышается вместе с нагрузкой, которую мы даём на кластер. Чтобы немного понизить high tail latencies, мы немного подтюнили JVM, на которой Stargate основан.
2. В конце мы автоматизировали это, чтобы мы могли ранить это в любой момент времени, как мы девелопим Stargate.

Client libraries

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

Каждый сервис определяет свой swagger spec каждый endpoint, который экспозится сервисом. Для нашего сервиса есть один API endpoint GraphQL/{keyspace}. То есть разработчики дают по сути keyspace и отправляют их GraphQL-запрос. Что происходит дальше: мы берём swagger spec и генерируем Client Libraries.

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

Вот пример, как клиент коннектился бы к нашему Stargate-клиенту. Ему надо просто зарепортить Client Library, вызвать GraphQL Query метод и отправить запрос.

Что это даёт:

1. Мы автоматически можем абстрагировать сложность Service Discovery (обратите внимание, другой сервис нигде не пишет host и port).
2. Мы имеем тонкий контроль над request budget. Например, говорим, что на какую-то страницу в Yelp максимально можем потратить 2 секунды. Запрос идёт между кучей микросервисов и мы можем сказать, что некие два микросервиса имеют бюджет 100 миллисекунды. Если мы превышаем этот бюджет, то отправлять таймаут-ответ.
3. Мы можем работать с distributed tracing инженеры могут следить, что происходит с их запросами.
4. Можем обеспечивать метрики прозрачности: количество запросов, количество запросов, которые отправили ответ со статусом 200 или другим.

GraphQL Playground

Playground это интерактивный development environment, который экспозится. Если у вас бэкенд на GraphQL, вы можете использовать Playground. Stargate делает это out of the box. По сути это такой IDE, которому вы просто говорите /graphqlschema, и вы можете в интерактивном режиме ранить запросы на Cassandra-кластер, быстро итерировать над своим GraphQL Queries.

Что он даёт: автоматически можно экспозить, автоматически открыть все возможные операции, которые можно сделать над этим кластером.

Operability

С точки зрения Operability, когда мы работали над Stargate, то было буквально несколько вещей, которые нам нужно было сделать так, чтобы включить его в нашу Yelp-инфраструктуру. Первое это Distributed Tracing Integration. Второе это то, что нужно заэкспортить какие-то метрики Stargate. Опять же, Load чтобы мы могли в реальном времени наблюдать, что происходит в кластере, смотреть CPU, память

Так как Stargate присоединяется к Cassandra-кластеру, то это даёт нам такую интересную property, чтобы мы могли заэкспортить Cassandra-метрики из этого кластера. Потом нам нужно было сделать немного изменений с точки зрения логинга (чтобы могли логить какие-то ошибки) и добавить TLS между нодами.