Zookeeper clickhouse что это
Clickhouse. Расширение кластера
Всем привет! Я хотел бы поделиться своим опытом по расширению высоконагруженного кластера ClickHouse, немного о том как работает репликация и шардирование.
Репликация
Репликация работает в рамках одного шарда на уровне таблиц, которые были созданы используя семейства движков ReplicatedMergeTree. На каждом шарде, репликация работает независимо от других шардов. Репликация в Clickhouse работает посредством Apache Zookeeper, где хранятся метаданные о репликах. Apache Zookeeper также отказоустойчив, и можно собрать ансамбль из 3 нод для отказоустойчивости.
Шардирование
Отказоустойчивость
Для нашей компании потеря данных недопустима, поэтому нужно обеспечить максимальную сохранность данных. Для отказоустойчивого кластера важно иметь нечетное количество узлов в рамках одного шарда. Считаю, что иметь в шарде 3 реплики достаточным, т.к. 3 реплики позволяет обеспечить «кворумную запись», также позволяет терпеть крах одного из реплик в шарде, не оказывая влияние на работу кластера.
Кворумная запись. Почему реплик в шарде 3 а не 2?
Предположим, что одна из реплик шарда приняла запись, обычно остальные реплики должны между собой асинхронно синхронизировать запись. Но, что если после принятия записи сервер не успел среплецировать принятую запись на остальные реплики и потерпел крах без возможности восстановления. В таком случае, данные будут потеряны. В ClickHouse, есть возможность задать «кворумную запись» при которой запись будет считаться успешной, если при записи в каждом шарде, запись приняли минимум 2 узла из 3. Если в шарде будут недоступны 2 узла из 3, то в таком случае кластер записи принимать не будет, и данные будут копиться в менеджере очередей MQ. При такой стратегии записи, стоит позаботиться об объёме хранилища менеджера очередей MQ, для того, чтобы было время привести реплики в рабочее состояние, иначе банально может не хватить места на диске для записи.
Файл конфигурации config.xml
У ClickHouse, есть файл конфигурации config.xml, в котором можно задать параметры для сервера ClickHouse. Для удобства из config.xml, я вывел отдельно параметры, которые задаются уникально для каждого отдельного узла. Параметры которые будут одинаковы на всех серверах, оставил их в config.xml. Ниже будут указаны примеры:
Расширение кластера
Настоятельно рекомендую использовать плейбуки при разворачивании новых нод, хорошо написанный плейбук, позволяет избегать ненужных ошибок и сэкономит ваше время.
После того как вы закончили разворачивание новых нод, нужно с рабочего кластера получить скрипт на создание таблиц c движками ReplicatedMerge и Distributed и при помощи этих скриптов создать таблицы на новых серверах.
Если вы выполните подобный команду:
и попытаетесь применить его на сервере CickHouse, то вы получите подобную ошибку:
Но если вы выполните команду таким образом:
у вас получится выполнить команду создания таблицы путем импорта скрипта:
После создания таблиц, нужно создать пользователей и дать привилегии этим пользователям которые будут обращаться к таблицам.
К сожалению в ClickHouse нельзя выполнив «show create user» получить команду на создания пользователя, команду наделения привилегий для пользователя и также получить хеш пароль пользователя, как это устроено в MySQL. Т.е. если вы забыли или не знаете пароль пользователя, то у вас не получится его создать также как на рабочем кластере.
Можно выделить важные моменты при расширении кластера:
Версии новых нод, должны быть такими же как с расширяемым кластером.
Все настройки указанные в users.xml, должны быть идентичны между собой. Иначе это может привести к непредвиденному поведению кластера.
После завершения разворачивания новых нод, нужно на каждом новом узле создать таблицы, которые имеются в кластере.
Т.к. в ClickHouse нельзя перешардировать данные в кластере в рамках одной таблицы. Чтобы увеличить приоритет новых записей на новые шарды, нужно перераспределить «weight». Не давайте слишком большой «weight» на новые шарды, если вы неуверены, что новые шарды способны выдержать оказываемую нагрузку на запись.
Создание пользователей и наделение привилегии.
Зачем нужно держать клетки в зоопарке закрытыми
В этой статье будет история об одной весьма характерной уязвимости в протоколе репликации в ClickHouse, а также будет показано, как можно расширить плоскость атаки.
ClickHouse — это база данных для хранения больших объемов данных, чаще всего используется больше одной реплики. Кластеризация и репликация в ClickHouse строятся поверх Apache ZooKeeper (ZK) и требуют прав на запись.
Установка ZK по умолчанию не требует аутентификации, так что тысячи ZK серверов, используемых для конфигурации Kafka, Hadoop, ClickHouse доступны публично.
Для сокращения плоскости атаки вы всегда должны настраивать аутентификацию и авторизацию при установке ZooKeeper
Есть конечно несколько 0day на основе Java десериализации, но представьте себе, что злоумышленник может читать и писать в ZooKeeper, используемый для репликации ClickHouse.
Например вы создаете узел /clickhouse/task_queue/ddl/query-0001 с содержимым:
а после этого на серверах кластера host1 и host2 таблица test будет удалена. DDL также поддерживает запуск запросов CREATE/ALTER/DROP.
Звучит страшно? Но где же атакующий сможет получить адреса серверов?
Репликация ClickHouse работает на уровне отдельных таблиц, так что при создании таблицы в ZK задается сервер, который будет отвечать за обмен метаданными с репликами. Например при выполнении запроса (ZK должен быть настроен, chXX — имя реплики, foobar — имя таблицы):
будут созданы узлы columns и metadata.
Содержимое /clickhouse/tables/01/foobar/replicas/chXX/hosts:
Можно ли слить данные из этого кластера? Да, если порт репликации ( TCP/9009 ) на сервере chXX-address не будет закрыт firewall и не будет настроена аутентификация для репликации. Как обойти аутентификацию?
Содержимое /clickhouse/tables/01–01/foobar/replicas/attacker/host:
Затем надо сказать остальным репликам, что на сервере атакующего есть новый блок данных, который им надо забрать — создается узел в ZK /clickhouse/tables/01-01/foobar/log/log-00000000XX (XX монотонно растущий счетчик, который должен быть больше, чем последний в журнале событий):
где source_replica — имя реплики атакующего, созданной на предыдущем шаге, block_id — идентификатор блока данных, get — команда «get block» (а тут команды для других операций).
Далее каждая реплика читает новое событие в журнале и идет на сервер, подконтрольный злоумышленнику, для получения блока данных (протокол репликации двоичный, работает поверх HTTP). Сервер attacker.com будет получать запросы:
где XXX — и есть данные аутентификации для репликации. В некоторых случаях это может быть учетная запись с доступом к базе данных по основному протоколу ClickHouse и протоколу HTTP. Как вы видели, плоскость атаки становится критически большой, потому что ZooKeeper, используемый для репликации, остался без настроенной аутентификации.
Давайте посмотрим на функцию получения блока данных из реплики, она написана с полной уверенностью, что все реплики под правильным контролем и между ними есть доверие.
код обработки репликации
Функция читает список файлов, затем их имена, размеры, содержимое, после чего пишет их в файловой системе. Стоит отдельно описать, как хранятся данные в файловой системе.
Есть несколько подкаталогов в /var/lib/clickhouse (каталог хранения по-умолчанию из конфигурационного файла):
flags — каталог для записи флагов, используемых при восстановлении после потери данных;
tmp — каталог хранения временных файлов;
user_files — операции с файлами в запросах ограничены этим каталогом (INTO OUTFILE и другие);
metadata — файлы sql с описаниями таблиц;
preprocessed_configs — обработанные производные конфигурационные файлы из /etc/clickhouse-server ;
data — собственно каталог с самими данными, в этм случае для каждой базы просто создается отдельный подкаталог здесь (например /var/lib/clickhouse/data/default ).
Для каждой таблицы создается подкаталог в каталоге с базой данных. Каждый столбец — отдельный файл в зависимости от формата движка. Например для таблицы foobar, созданной атакующим, будут созданы следующие файлы:
Реплика ожидает получения файлов с такими же именами при обработке блока данных и не проверяет их каким-либо способом.
Есть несколько вариантов превращения возможности записи файлов в удаленный запуск кода (RCE).
Внешние словари в RCE
ODBC в RCE
Давайте создадим файл
/.odbc.ini с таким содержимым:
затем при запуске SELECT * FROM odbc(‘DSN=lalala’, ‘test’, ‘test’); будет подгружена библиотека test.so и получено RCE (спасибо buglloc за наводку).
Эти и другие уязвимости были исправлены в версии ClickHouse 19.14.3. Берегите свои ClickHouse и ZooKeepers!
Репликация данных
Репликация поддерживается только для таблиц семейства MergeTree:
Репликация работает на уровне отдельных таблиц, а не всего сервера. То есть, на сервере могут быть расположены одновременно реплицируемые и не реплицируемые таблицы.
Репликация не зависит от шардирования. На каждом шарде репликация работает независимо.
ClickHouse хранит метаинформацию о репликах в Apache ZooKeeper. Используйте ZooKeeper 3.4.5 или новее.
Для использовании репликации, установите параметры в секции zookeeper конфигурации сервера.
Не пренебрегайте настройками безопасности. ClickHouse поддерживает ACL схему digest подсистемы безопасности ZooKeeper.
Пример указания адресов кластера ZooKeeper:
Если в конфигурационном файле не настроен ZooKeeper, то вы не сможете создать реплицируемые таблицы, а уже имеющиеся реплицируемые таблицы будут доступны в режиме только на чтение.
Для очень больших кластеров, можно использовать разные кластеры ZooKeeper для разных шардов. Впрочем, на кластере Яндекс.Метрики (примерно 300 серверов) такой необходимости не возникает.
Репликация асинхронная, мульти-мастер. Запросы INSERT и ALTER можно направлять на любой доступный сервер. Данные вставятся на сервер, где выполнен запрос, а затем скопируются на остальные серверы. В связи с асинхронностью, только что вставленные данные появляются на остальных репликах с небольшой задержкой. Если часть реплик недоступна, данные на них запишутся тогда, когда они станут доступны. Если реплика доступна, то задержка составляет столько времени, сколько требуется для передачи блока сжатых данных по сети. Количество потоков для выполнения фоновых задач можно задать с помощью настройки background_schedule_pool_size.
Движок ReplicatedMergeTree использует отдельный пул потоков для скачивания кусков данных. Размер пула ограничен настройкой background_fetches_pool_size, которую можно указать при перезапуске сервера.
Каждый блок данных записывается атомарно. Запрос INSERT разбивается на блоки данных размером до max_insert_block_size = 1048576 строк. То есть, если в запросе INSERT менее 1048576 строк, то он делается атомарно.
При репликации, по сети передаются только исходные вставляемые данные. Дальнейшие преобразования данных (слияния) координируются и делаются на всех репликах одинаковым образом. За счёт этого минимизируется использование сети, и благодаря этому, репликация хорошо работает при расположении реплик в разных дата-центрах. (Стоит заметить, что дублирование данных в разных дата-центрах, по сути, является основной задачей репликации).
Количество реплик одних и тех же данных может быть произвольным. В Яндекс.Метрике в продакшене используется двукратная репликация. На каждом сервере используется RAID-5 или RAID-6, в некоторых случаях RAID-10. Это является сравнительно надёжным и удобным для эксплуатации решением.
Система следит за синхронностью данных на репликах и умеет восстанавливаться после сбоя. Восстановление после сбоя автоматическое (в случае небольших различий в данных) или полуавтоматическое (когда данные отличаются слишком сильно, что может свидетельствовать об ошибке конфигурации).
Создание реплицируемых таблиц
Параметры Replicated*MergeTree
Как видно в примере, эти параметры могут содержать подстановки в фигурных скобках. Эти подстановки заменяются на соответствующие значения из конфигурационного файла, из секции macros.
Путь к таблице в ZooKeeper должен быть разным для каждой реплицируемой таблицы. В том числе, для таблиц на разных шардах, должны быть разные пути.
В данном случае, путь состоит из следующих частей:
/clickhouse/tables/ — общий префикс. Рекомендуется использовать именно его.
Имя реплики — то, что идентифицирует разные реплики одной и той же таблицы. Можно использовать для него имя сервера, как показано в примере. Впрочем, достаточно, чтобы имя было уникально лишь в пределах каждого шарда.
Можно не использовать подстановки, а указать соответствующие параметры явно. Это может быть удобным для тестирования и при настройке маленьких кластеров. Однако в этом случае нельзя пользоваться распределенными DDL-запросами ( ON CLUSTER ).
При работе с большими кластерами мы рекомендуем использовать подстановки, они уменьшают вероятность ошибки.
Можно указать аргументы по умолчанию для движка реплицируемых таблиц в файле конфигурации сервера.
В этом случае можно опустить аргументы при создании таблиц:
Это будет эквивалентно следующему запросу:
Выполните запрос CREATE TABLE на каждой реплике. Запрос создаёт новую реплицируемую таблицу, или добавляет новую реплику к имеющимся.
Если вы добавляете новую реплику после того, как таблица на других репликах уже содержит некоторые данные, то после выполнения запроса, данные на новую реплику будут скачаны с других реплик. То есть, новая реплика синхронизирует себя с остальными.
Восстановление после сбоя
Если при старте сервера, недоступен ZooKeeper, реплицируемые таблицы переходят в режим только для чтения. Система будет пытаться периодически установить соединение с ZooKeeper.
Если при INSERT недоступен ZooKeeper, или происходит ошибка при взаимодействии с ним, будет выкинуто исключение.
При подключении к ZooKeeper, система проверяет соответствие между имеющимся в локальной файловой системе набором данных и ожидаемым набором данных (информация о котором хранится в ZooKeeper). Если имеются небольшие несоответствия, то система устраняет их, синхронизируя данные с реплик.
Обнаруженные битые куски данных (с файлами несоответствующего размера) или неизвестные куски (куски, записанные в файловую систему, но информация о которых не была записана в ZooKeeper) переносятся в поддиректорию detached (не удаляются). Недостающие куски скачиваются с реплик.
Стоит заметить, что ClickHouse не делает самостоятельно никаких деструктивных действий типа автоматического удаления большого количества данных.
Для запуска восстановления, создайте в ZooKeeper узел /path_to_table/replica_name/flags/force_restore_data с любым содержимым или выполните команду для восстановления всех реплицируемых таблиц:
Затем запустите сервер. При старте, сервер удалит эти флаги и запустит восстановление.
Восстановление в случае потери всех данных
Если на одном из серверов исчезли все данные и метаданные, восстановление делается следующим образом:
Затем запустите сервер (перезапустите, если уже запущен). Данные будут скачаны с реплик.
Отсутствует ограничение на использование сетевой полосы при восстановлении. Имейте это ввиду, если восстанавливаете сразу много реплик.
Преобразование из MergeTree в ReplicatedMergeTree
Если на разных репликах данные отличаются, то сначала синхронизируйте их, либо удалите эти данные на всех репликах кроме одной.
Преобразование из ReplicatedMergeTree в MergeTree
Восстановление в случае потери или повреждения метаданных на ZooKeeper кластере
Если данные в ZooKeeper оказались утеряны или повреждены, то вы можете сохранить данные, переместив их в нереплицируемую таблицу, как описано в пункте выше.
Масштабирование ClickHouse, управление миграциями и отправка запросов из PHP в кластер
В предыдущей статье мы поделились своим опытом внедрения и использования СУБД ClickHouse в компании СМИ2. В текущей статье мы затронем вопросы масштабирования, которые возникают с увеличением объема анализируемых данных и ростом нагрузки, когда данные уже не могут храниться и обрабатываться в рамках одного физического сервера. Также мы расскажем о разработанном нами инструменте для миграции DDL-запросов в ClickHouse-кластер.
ClickHouse специально проектировался для работы в кластерах, расположенных в разных дата-центрах. Масштабируется СУБД линейно до сотен узлов. Так, например, Яндекс.Метрика на момент написания статьи — это кластер из более чем 400 узлов.
ClickHouse предоставляет шардирование и репликацию «из коробки», они могут гибко настраиваться отдельно для каждой таблицы. Для обеспечения реплицирования требуется Apache ZooKeeper (рекомендуется использовать версию 3.4.5+). Для более высокой надежности мы используем ZK-кластер (ансамбль) из 5 узлов. Следует выбирать нечетное число ZK-узлов (например, 3 или 5), чтобы обеспечить кворум. Также отметим, что ZK не используется в операциях SELECT, а применяется, например, в ALTER-запросах для изменений столбцов, сохраняя инструкции для каждой из реплик.
Шардирование
Шардирование в ClickHouse позволяет записывать и хранить порции данных в кластере распределенно и обрабатывать (читать) данные параллельно на всех узлах кластера, увеличивая throughput и уменьшая latency. Например, в запросах с GROUP BY ClickHouse выполнит агрегирование на удаленных узлах и передаст узлу-инициатору запроса промежуточные состояния агрегатных функций, где они будут доагрегированы.
Для шардирования используется специальный движок Distributed, который не хранит данные, а делегирует SELECT-запросы на шардированные таблицы (таблицы, содержащие порции данных) с последующей обработкой полученных данных. Запись данных в шарды может выполняться в двух режимах: 1) через Distributed-таблицу и необязательный ключ шардирования или 2) непосредственно в шардированные таблицы, из которых далее данные будут читаться через Distributed-таблицу. Рассмотрим эти режимы более подробно.
В первом режиме данные записываются в Distributed-таблицу по ключу шардирования. В простейшем случае ключом шардирования может быть случайное число, т. е. результат вызова функции rand(). Однако в качестве ключа шардирования рекомендуется брать значение хеш-функции от поля в таблице, которое позволит, с одной стороны, локализовать небольшие наборы данных на одном шарде, а с другой — обеспечит достаточно ровное распределение таких наборов по разным шардам в кластере. Например, идентификатор сессии (sess_id) пользователя позволит локализовать показы страниц одному пользователю на одном шарде, при этом сессии разных пользователей будут распределены равномерно по всем шардам в кластере (при условии, что значения поля sess_id будут иметь хорошее распределение). Ключ шардирования может быть также нечисловым или составным. В этом случае можно использовать встроенную хеширующую функцию cityHash64. В рассматриваемом режиме данные, записываемые на один из узлов кластера, по ключу шардирования будут перенаправляться на нужные шарды автоматически, увеличивая, однако, при этом трафик.
Более сложный способ заключается в том, чтобы вне ClickHouse вычислять нужный шард и выполнять запись напрямую в шардированную таблицу. Сложность здесь обусловлена тем, что нужно знать набор доступных узлов-шардов. Однако в этом случае запись становится более эффективной, и механизм шардирования (определения нужного шарда) может быть более гибким.
Репликация
ClickHouse поддерживает репликацию данных, обеспечивая целостность данных на репликах. Для репликации данных используются специальные движки MergeTree-семейства:
Репликация часто применяется вместе с шардированием. Например, кластер из 6 узлов может содержать 3 шарда по 2 реплики. Следует отметить, что репликация не зависит от механизмов шардирования и работает на уровне отдельных таблиц.
Запись данных может выполняться в любую из таблиц-реплик, ClickHouse выполняет автоматическую синхронизацию данных между всеми репликами.
Примеры конфигурации ClickHouse-кластера
Один шард и четыре реплики
Пример схемы создания таблицы:
Пример SQL-запроса создания таблицы для указанной конфигурации:
Преимущество данной конфигурации:
Четыре шарда по одной реплике
Пример SQL-запроса создания таблицы для указанной конфигурации:
Преимущество данной конфигурации:
Два шарда по две реплики
Пример SQL-запроса создания таблицы для указанной конфигурации:
Данная конфигурация воплощает лучшие качества из первого и второго примеров:
Пример конфигурации кластеров в ansible
Конфигурация кластеров в ansible может выглядеть следующим образом:
PHP-драйвер для работы с ClickHouse-кластером
В предыдущей статье мы уже рассказывали о нашем open-source PHP-драйвере для ClickHouse.
Когда количество узлов становится большим, управление кластером становится неудобным. Поэтому мы разработали простой и достаточно функциональный инструмент для миграции DDL-запросов в ClickHouse-кластер. Далее мы кратко опишем на примерах его возможности.
Для подключения к кластеру используется класс ClickHouseDB\Cluster :
Драйвер выполняет подключение к кластеру и отправляет ping-запросы на каждый узел, перечисленный в DNS-записи.
Установка максимального времени подключения ко всем узлам кластера настраивается следующим образом:
Проверка состояния реплик кластера выполняется так:
Состояние ClickHouse-кластера проверяется следующим образом:
Для облегченной проверки в драйвере необходимо установить специальный флаг:
Получение списка всех доступных кластеров делается следующим образом:
Например, получить конфигурацию кластеров, которые были описаны выше, можно так:
Получение списка узлов по названию кластера или из шардированных таблиц:
Получение размера таблицы или размеров всех таблиц через отправку запроса на каждый узел кластера:
Получение списка таблиц кластера:
Определение лидера в кластере:
Очистка данных в таблице в кластере:
Инструмент миграции DDL-запросов
Для миграции DDL-запросов для реляционных СУБД в нашей компании используется MyBatis Migrations.
Об инструментах миграции на Хабре уже писали:
Для работы с ClickHouse-кластером нам требовался аналогичный инструмент.
На момент написания статьи ClickHouse имеет ряд особенностей (ограничений) связанных с DDL-запросами. Цитата:
Реплицируются INSERT, ALTER (см. подробности в описании запроса ALTER). Реплицируются сжатые данные, а не тексты запросов. Запросы CREATE, DROP, ATTACH, DETACH, RENAME не реплицируются — то есть, относятся к одному серверу. Запрос CREATE TABLE создаёт новую реплицируемую таблицу на том сервере, где выполняется запрос; а если на других серверах такая таблица уже есть — добавляет новую реплику. Запрос DROP TABLE удаляет реплику, расположенную на том сервере, где выполняется запрос. Запрос RENAME переименовывает таблицу на одной из реплик — то есть, реплицируемые таблицы на разных репликах могут называться по разному.
Команда разработчиков ClickHouse уже анонсировала работу в этом направлении, но в настоящее время приходится решать эту задачу внешним инструментарием. Мы создали простой прототип инструмента phpMigrationsClickhouse для миграции DDL-запросов в ClickHouse-кластер. И в наших планах — абстрагировать phpMigrationsClickhouse от языка PHP.
Опишем алгоритм, использующийся в настоящий момент в phpMigrationsClickhouse, который может быть реализован на любом другом языке программирования.
На текущий момент инструкция по миграции в phpMigrationsClickhouse состоит из:
Создадим PHP-файл, содержащий следующий код:
Добавим SQL-запросы, которые нужно накатить:
Добавим SQL-запросы для выполнения отката в случае ошибки:
Существует 2 стратегии накатывания миграций:
При возникновении ошибки возможны следующие варианты:
Отдельное место занимают ошибки, когда не известно состояние кластера:
Принцип работы PHP-кода при выполнении миграции следующий:
В случае ошибки выполняется отправка на все узлы кластера downgrade-запроса:
Мы продолжим цикл материалов, посвященных нашему опыту работы с ClickHouse.
В завершение статьи мы хотели бы провести небольшой опрос.