З Elasticsearch зустрічаються багато. Але що відбувається, коли хочеш за його допомогою зберігати логи «в особливо великому обсязі»? Та ще й безболісно переживати відмову будь-якого з кількох дата-центрів? Який варто робити архітектуру, і на яке підводне каміння натрапиш?
Ми в Однокласниках вирішили за допомогою elasticsearch вирішити питання лог-менеджменту, а тепер ділимося з Хабром досвідом: і про архітектуру, і про підводні камені.
Я – Петро Зайцев, працюю системним адміністратором в Однокласниках. До цього теж був адміном, працював із Manticore Search, Sphinx search, Elasticsearch. Можливо, якщо з'явиться ще якийсь …search, ймовірно працюватиму і з ним. Також беру участь у низці опенсорсних проектів на добровільній основі.
Коли я прийшов до Однокласників, то необачно сказав на співбесіді, що вмію працювати з Elasticsearch. Після того, як освоївся і зробив якісь простенькі завдання, мені подарували велике завдання щодо реформування системи лог-менеджменту, яка існувала на той момент.
Вимоги
Вимоги до системи були сформульовані так:
- Як фронтенд повинен був використовуватися Graylog. Тому що в компанії вже був досвід використання цього продукту, програмісти та тестувальники його знали, він був звичний і зручний.
- Обсяг даних: в середньому 50-80 тисяч повідомлень за секунду, але якщо щось ламається, то трафік нічим не обмежений, це може бути 2-3 мільйони рядків за секунду
- Обговоривши із замовниками вимоги щодо швидкості обробки пошукових запитів, ми зрозуміли, що типовий патерн використання такої системи такий: люди шукають логи свого додатку за останні два дні і не хочуть чекати на результат сформульованого запиту більше секунди.
- Адміни наполягали на тому, щоб система за необхідності легко масштабувалася, не вимагаючи від них глибокого вникання у те, як вона влаштована.
- Щоб єдине завдання з обслуговування, яке цим системам потрібно періодично — це міняти якесь залізо.
- Крім того, в Однокласниках є чудова технічна традиція: будь-який сервіс, який ми запускаємо, має переживати відмову дата-центру (раптова, незапланована і абсолютно в будь-який час).
Остання вимога щодо реалізації цього проекту далася нам найбільшою кров'ю, про що я ще розповім докладніше.
Середа
Ми працюємо на чотирьох дата-центрах, при цьому дата-ноди Elasticsearch можуть розташовуватися тільки в трьох (по ряду нетехнічних причин).
У цих чотирьох дата-центрах знаходяться приблизно 18 тисяч джерел лігів — залізниці, контейнери, віртуальні машини.
Важлива особливість: запуск кластера відбувається у контейнерах не на фізичних машинах, а на . Контейнерам гарантується 2 ядра, аналогічні 2.0Ghz v4 з можливістю утилізації інших ядер, у разі їх простою.
Іншими словами:
топологія
Загальний вид рішення мені спочатку бачився так:
- 3-4 VIP стоять за А-рекордом домену Graylog, це адреса, на яку надсилаються логи.
- кожен VIP є балансувальником LVS.
- Після нього логи потрапляють на батарею Graylog, частина даних у форматі GELF, частина у форматі syslog.
- Далі все це великими батчами пишеться в батарею з координаторів Elasticsearch.
- А вони, своєю чергою, надсилають запити на запис та читання на релевантні дата-ноди.
Термінологія
Можливо, не всі докладно знаються на термінології, тому хотілося б трохи зупинитися на ній.
У Elasticsearch є кілька типів нод - master, coordinator, data node. Є ще два інших типи для різних перетворень логів та зв'язку різних кластерів між собою, але ми використовували лише перелічені.
Майстер
Пінгує всі присутні в кластері ноди, підтримує актуальну карту кластера і поширює її між нодами, обробляє подієву логіку, займається різного роду cluster wide housekeeping.
Координатор
Виконує одне-єдине завдання: приймає запити від клієнтів на читання чи запис та маршрутизує цей трафік. Якщо запит на запис, швидше за все, він запитає master, в який шард релевантного індексу йому це покласти, і перенаправить запит далі.
Data node
Зберігає дані, виконує пошукові запити та операції над розташованими на ній шардами.
Сірий
Це щось схоже на сплав Kibana з Logstash в ELK-стеку. Graylog поєднує в собі і UI та конвеєр з обробки логів. Під капотом у Graylog працюють Kafka та Zookeeper, які забезпечують зв'язність Graylog як кластера. Graylog вміє кешувати логи (Kafka) на випадок недоступності Elasticsearch і повторювати невдалі запити на читання та запис, групувати та маркувати за правилами логи. Як і Logstash, Graylog має функціональність модифікації рядків перед записом в Elasticsearch.
Крім того, у Graylog є вбудований service discovery, що дозволяє на основі однієї доступної ноди Elasticsearch отримати всю карту кластера і відфільтрувати її за певним тегом, що дає можливість надсилати запити на певні контейнери.
Візуально це виглядає приблизно так:
Це скріншот із конкретної інстансу. Тут ми пошуковим запитом вибудовуємо гістограму, виводимо релевантні рядки.
Індекси
Повертаючись до архітектури системи, я хотів би детальніше зупинитися на тому, як ми будували модель індексів, щоб усе це працювало коректно.
На наведеній раніше схемі це найнижчий рівень: Elasticsearch data nodes.
Індекс - це велика віртуальна сутність, що складається з шардів Elasticsearch. Сам по собі кожен із шардів є нічим іншим, як Lucene index. А кожен Lucene index, у свою чергу, складається і з одного або більше сегментів.
При проектуванні ми прикидали, що для забезпечення вимоги швидкості читання на великому обсязі даних нам необхідно рівномірно «розмазати» ці дані по дата-нодам.
Це вилилося в те, що кількість шардів на індекс (з репліками) у нас має бути строго дорівнює кількості дата-нод. По-перше, для того, щоб забезпечити replication factor, рівний двом (тобто ми можемо втратити половину кластера). А по-друге, для того, щоб запити на читання та запис обробляти, як мінімум, на половині кластера.
Час зберігання ми визначили спочатку як 30 днів.
Розподіл шардів можна подати графічно в такий спосіб:
Весь темно-сірий прямокутник цілком індекс. Лівий червоний квадрат у ньому це primary-шард, перший в індексі. А блакитний квадрат – це replica-шард. Вони знаходяться у різних дата-центрах.
Коли ми додаємо ще один шард, він потрапляє до третього дата-центру. І, зрештою, ми отримуємо таку структуру, яка забезпечує можливість втрати ДЦ без втрати консистентності даних:
Ротацію індексів, тобто. створення нового індексу та видалення найстарішого, ми зробили рівною 48 годин (за патерном використання індексу: за останніми 48 годинами шукають найчастіше).
Такий інтервал ротації індексів пов'язаний із такими причинами:
Коли на конкретну дата-ноду прилітає пошуковий запит, то, з погляду перформансу вигідніше, коли опитується один шард, якщо його розмір можна порівняти з розміром хіпа ноди. Це дозволяє тримати гарячу частину індексу в хіпі і швидко до неї звертатися. Коли "гарячих частин" стає багато, то деградує швидкість пошуку за індексом.
Коли нода починає виконувати пошуковий запит на одному шарді, вона виділяє кількість тредів, що дорівнює кількості гіпертредінгових ядер фізичної машини. Якщо пошуковий запит зачіпає велику кількість шардів, то кількість тредів зростає пропорційно. Це погано відбивається на швидкості пошуку та негативно позначається на індексації нових даних.
Щоб забезпечити необхідний пошук latency, ми вирішили використовувати SSD. Для швидкої обробки запитів машини, на яких розміщувалися ці контейнери, повинні були мати щонайменше 56 ядр. Цифра в 56 вибирається як умовно-достатня величина, що визначає кількість тредів, які породжують еластичнідослідження в процесі роботи. У Elasitcsearch багато параметрів thread pool безпосередньо залежать від кількості доступних ядер, що в свою чергу прямо впливає на необхідну кількість нід в кластері за принципом "менше ядер - більше нод".
У результаті у нас вийшло, що в середньому шард важить десь 20 гігабайт, і на 1 індекс припадає 360 шардів. Відповідно, якщо ми їх ротуємо раз на 48 годин, то у нас їх 15 штук. Кожен індекс вміщує дані за 2 дні.
Схеми запису та читання даних
Давайте розберемося, як у цій системі записуються дані.
Припустимо, у нас із Graylog прилітає до координатора якийсь запит. Наприклад, ми хочемо проіндексувати 2-3 тисяч рядків.
Координатор, отримавши від Graylog запит, опитує майстер: "У запиті на індексацію у нас був конкретно вказаний індекс, але в який шард це писати - не вказано".
Master відповідає: «Запиши цю інформацію в шард номер 71», після чого вона направляється безпосередньо в релевантну дату-ноду, де знаходиться primary-shard номер 71.
Після цього лог транзакцій реплікується на replica-shard, який знаходиться вже в іншому дата-центрі.
З Graylog до координатора прилітає пошуковий запит. Координатор перенаправляє його за індексом, при цьому Elasticsearch за принципом round-robin розподіляє запити між primary-shard і replica-shard.
Ноди в кількості 180 штук відповідають нерівномірно, і поки вони відповідають, координатор накопичує інформацію, яку в нього вже «виплюнули» швидші дата-ноди. Після цього, коли або вся інформація прийшла, або на запит досягнуть тайм-аут, віддає все безпосередньо клієнту.
Вся ця система в середньому відпрацьовує пошукові запити за останні 48 годин за 300-400ms, виключаючи ті запити, які з leading wildcard.
"Квіточки" з Elasticsearch: налаштування Java
Щоб усе це запрацювало так, як ми спочатку хотіли, ми дуже довго налагоджували найрізноманітніші речі у кластері.
Перша частина виявлених проблем була пов'язана з тим, як в Elasticsearch по дефолту налаштована Java.
Проблема перша
Ми спостерігали дуже велику кількість повідомлень про те, що у нас на рівні Lucene, коли запущені background job'и, мерджі сегментів Lucene завершуються з помилкою. При цьому в логах було видно, що це OutOfMemoryError-помилка. По телеметрії ми бачили, що хіп вільний, і було зрозуміло, чому ця операція падає.
З'ясувалося, що Мерджі Lucene-індексів відбуваються поза хіпом. А контейнери досить жорстко обмежені за ресурсами, що споживаються. У ці ресурси влазив лише хіп (значення heap.size було приблизно рівне RAM), а якісь off-heap операції падали з помилкою алокації пам'яті, якщо з якоїсь причини не вкладалися в ті ~500MB, що залишалися до ліміту.
Фікс був досить тривіальним: доступний для контейнера обсяг RAM збільшили, після чого забули, що такі проблеми у нас взагалі були.
Проблема друга
Дня через 4-5 після запуску кластера ми помітили, що дата-ноди починають періодично вивалюватися з кластера і заходити до нього секунд через 10-20.
Коли полізли розбиратися, з'ясувалося, що ця сама off-heap пам'ять у Elasticsearch не контролюється практично ніяк. Коли ми контейнеру віддали більше пам'яті, ми отримали можливість заповнювати direct buffer pools різною інформацією, і вона очищалася тільки після того, як запускався explicit GC з боку Elasticsearch.
У ряді випадків ця операція відбувалася досить-таки довго, і за цей час кластер встигав помітити цю ноду, що вже вийшла. Ця проблема добре описана .
Рішення було наступним: ми обмежили Java можливість використовувати основну частину пам'яті поза хіпом під ці операції. Ми лімітували її до 16 гігабайт (-XX: MaxDirectMemorySize=16g), домігшись того, що explicit GC викликався значно частіше, а відпрацьовував значно швидше, переставши тим самим дестабілізувати кластер.
Проблема третя
Якщо ви думаєте, що проблеми з «нодами, що залишають кластер у найнепередбачуваніший момент» на цьому скінчилися, ви помиляєтеся.
Коли ми конфігурували роботу з індексами, ми зупинили свій вибір на mmapfs, щоб по свіжих шардах з великою сегментованістю. Це було досить грубою помилкою, тому що при використанні mmapfs файл маппіт в оперативну пам'ять, а далі ми працюємо вже з mapped-файлом. Через це виходить, що при спробі GC зупинити треди в додатку ми дуже довго йдемо в safepoint, і по дорозі до нього додаток перестає відповідати на запити майстра про те, чи воно живе. Відповідно, master вважає, що нода у нас більше в кластері немає. Після цього через 5-10 секунд відпрацьовує garbage collector, нода оживає, знову заходить у кластер і починає ініціалізацію шардів. Все це сильно нагадувало "продакшен, який ми заслужили" і не годилося для чогось серйозного.
Щоб позбутися такої поведінки, ми спершу перейшли на стандартні niofs, а після, коли з п'ятих версій Elastic відмігрувалися на шости, спробували hybridfs, де ця проблема не відтворювалася. Детальніше про типи стореджа можна почитати .
Проблема четверта
Потім була ще дуже цікава проблема, яку ми лікували рекордно довго. Ми ловили її 2-3 місяці, тому що був абсолютно незрозумілий її патерн.
Іноді у нас координатори йшли у Full GC, зазвичай десь по обіді, і звідти вже не поверталися. При цьому при логуванні затримки GC це виглядало так: у нас все йде добре, добре, добре, а потім раз і все різко погано.
Спочатку ми думали, що у нас є злий користувач, який запускає якийсь запит, що вибиває координатор із робочого режиму. Дуже довго логували запити, намагаючись з'ясувати, що відбувається.
У результаті з'ясувалося, що в той момент, коли якийсь користувач запускає величезний запит, і він потрапляє на якийсь конкретний координатор Elasticsearch, деякі ноди відповідають довше, ніж інші.
І той час, поки координатор чекає відповіді всіх нод, він накопичує в собі результати, надіслані з нод, що вже відповіли. Для GC це означає, що ми дуже швидко змінюється патерн використання хіпа. І той GC, який ми використовували, із цим завданням не справлявся.
Єдиний фікс, який ми знайшли для того, щоб змінити поведінку кластера в такій ситуації – міграція на JDK13 та використання збирача сміття Shenandoah. Це вирішило проблему, координатори у нас перестали падати.
На цьому проблеми з Java закінчилися і почалися проблеми з пропускною здатністю.
"Ягідки" з Elasticsearch: пропускна спроможність
Проблеми з пропускною здатністю означають, що наш кластер працює стабільно, але на піках кількості індексованих документів і в момент маневрів продуктивність недостатня.
Перший зустрінутий симптом: при якихось «вибухах» на продакшені, коли різко генерується дуже багато логів, в Graylog починає часто мелькати помилка індексації es_rejected_execution.
Це відбувалося через те, що thread_pool.write.queue на одній дата-ноді до того моменту, як Elasticsearch зможе обробити запит на індексацію і закинути інформацію в шард на диск, по дефолту вміє кешувати лише 200 запитів. І в Про цей параметр говориться вкрай мало. Вказується лише гранична кількість тредів і дефолтний розмір.
Зрозуміло, ми пішли крутити це значення і з'ясували таке: саме в нашому сетапі досить добре кешується до 300 запитів, а більше значення загрожує тим, що ми знову відлітаємо в Full GC.
Крім того, оскільки це пачки повідомлень, які прилітають у рамках одного запиту, то треба було ще підкрутити Graylog для того, щоб він писав не часто і дрібними батчами, а величезними батчами або раз на 3 секунди, якщо батч все ще не сповнений. У такому разі виходить, що інформація, яку ми в Elasticsearch пишемо, стає доступною не через дві секунди, а через п'ять (що нас цілком влаштовує), але зменшується кількість ретраїв, які доводиться зробити, щоб пропхати велику пачку інформації.
Це особливо важливо в ті моменти, коли в нас щось десь впало і люто про це повідомляє, щоб не отримувати повністю заспамлений Elastic, а через якийсь час — непрацездатні через буфери, що забилися, ноди Graylog.
Крім того, коли у нас відбувалися ці вибухи на продакшені, у нас приходили скарги від програмістів і тестувальників: у той момент, коли їм дуже потрібні ці логи, вони видаються їм дуже повільно.
Почали розбиратися. З одного боку, було зрозуміло, що і пошукові запити, і запити на індексацію відпрацьовують, по суті, на тих самих фізичних машинах, і так чи інакше просадки будуть.
Але це можна було частково обійти за рахунок того, що в шостих версіях Elasticsearch з'явився алгоритм, що дозволяє розподіляти запити між релевантними дата-нодами не за випадковим принципом round-robin (контейнер, який займається індексацією та тримає primary-shard, може бути дуже зайнятий, там не буде можливості відповісти швидко), а направити цей запит на менш завантажений контейнер із replica-shard, який відповість значно швидше. Іншими словами, ми дійшли use_adaptive_replica_selection: true.
Картина читання починає виглядати так:
Перехід на цей алгоритм дозволив помітно покращити query time у ті моменти, коли у нас йшов великий потік логів на запис.
Зрештою, основна проблема полягала у безболісному виведенні дата-центру.
Чого ми хотіли від кластера відразу після втрати зв'язку з одним ДЦ:
- Якщо у нас в дата-центрі, що відвалився, знаходиться поточний master, то він буде переобраний і переїде як роль на іншу ноду в іншому ДЦ.
- Майстер швидко викине із кластера всі недоступні ноди.
- На основі решти він зрозуміє: в дата-центрі, що втратився, у нас були такі-то primary-шарди, швидко запромоутить компліментарні replica-шарди в дата-центрах, що залишилися, і у нас продовжиться індексація даних.
- В результаті цього у нас буде плавно деградувати пропускну здатність кластера на запис та читання, проте в цілому все працюватиме хоч і повільно, але стабільно.
Як з'ясувалося, хотіли ми чогось такого:
А отримали таке:
Як так вийшло?
У момент падіння дата-центру у нас вузьким місцем став майстер.
Чому?
Справа в тому, що в майстрі є TaskBatcher, який відповідає за поширення в кластері певних завдань, івентів. Будь-який вихід ноди, будь-яке просування шарда з replica в primary, будь-яке завдання створення десь якогось шарда — усе це потрапляє спочатку в TaskBatcher, де обробляється послідовно й у один потік.
У момент виведення одного дата-центру виходило, що всі дата-ноди в дата-центрах, що вижили, вважали своїм обов'язком повідомити майстру «у нас загубилися такі шарди і такі дата-ноди».
При цьому дата-ноди, що вижили, посилали всю цю інформацію поточному майстру і намагалися дочекатися підтвердження, що він її прийняв. Цього вони не чекали, оскільки майстер отримував завдання швидше, ніж встиг відповідати. Ноди по таймауту повторювали запити, а майстер у цей час навіть не намагався відповідати ними, а був повністю поглинений завданням сортування запитів по пріоритетності.
У термінальному вигляді виходило, що дата-ноди спамили майстра до того, що він йшов у full GC. Після цього у нас роль майстра переїжджала на якусь наступну ноду, з нею відбувалося те саме, і в результаті кластер розвалювався повністю.
Ми робили виміри, і до версії 6.4.0, де це було зафіксовано, нам було достатньо вивести одночасно вивести лише 10 дата-од з 360 для того, щоб повністю покласти кластер.
Виглядало це приблизно так:
Після версії 6.4.0, де полагодили цей стрімкий баг, дата-ноди перестали вбивати майстра. Але «розумнішим» він від цього не став. А саме: коли ми виводимо 2, 3 або 10 (будь-яка кількість, відмінна від одиниці) дата-нод, майстер отримує якесь перше повідомлення, яке говорить, що нода А вийшла, і намагається розповісти про цю ноду B, ноду C, ноді D.
І на даний момент з цим можна боротися тільки встановленням тайм-ауту на спроби комусь про щось розповісти, рівні десь 20-30 секунд, і таким чином керувати швидкістю виведення дата-центру з кластера.
В принципі це укладається в ті вимоги, які спочатку були пред'явлені до кінцевого продукту в рамках проекту, але з точки зору «чистої науки» це баг. Який, до речі, був успішно зафіксований розробниками у версії 7.2.
Причому, коли якась дата-нода виходила, виходило, що поширити інформацію про її вихід важливіше, ніж розповісти всьому кластеру, що на ній знаходилися такі primary-shard (щоб запромоутувати replica-shard в іншому дата-центрі в primary, і в них можна було писати інформацію).
Тому, коли вже все «відгриміло», дата-ноди, що вийшли, не маркуються як stale негайно. Відповідно, ми змушені чекати, коли відтаймуть всі пінги до вийшли дата-нод і тільки після цього наш кластер починає розповідати про те, що там, там і там треба продовжити запис інформації. Докладніше можна почитати про це .
У результаті операція виведення дата-центру у нас сьогодні займає близько 5 хвилин на пік. Для настільки великої та неповороткої махини це досить добрий результат.
У результаті ми дійшли наступного рішення:
- У нас 360 дата-нод із дисками на 700 гігабайт.
- 60 координаторів для роутингу трафіку за цими дата-нодами.
- 40 майстрів, які у нас залишилися як певна спадщина з часів версій до 6.4.0 — щоб пережити висновок дата-центру, ми морально були готові втратити кілька машин, щоб гарантовано навіть за найгіршого сценарію мати кворум майстрів
- Будь-які спроби поєднання ролей на одному контейнері в нас упиралися в те, що рано чи пізно нода ламалася під навантаженням.
- У всьому кластері використовується heap.size, що дорівнює 31 гігабайту: всі спроби зменшити розмір приводили до того, що на важких пошукових запитах з leading wildcard або вбивав якісь ноди, або прибивався circuit breaker в самому Elasticsearch.
- Крім того, для забезпечення продуктивності пошуку ми намагалися тримати кількість об'єктів у кластері мінімально можливою, щоб обробляти якнайменше подій у найвужчому місці, яке у нас вийшло у майстрі.
Насамкінець про моніторинг
Щоб усе це працювало так, як думалося, ми моніторимо наступне:
- Кожна дата-нода повідомляє в нашу хмару, що вона є, і на ній знаходяться такі шарди. Коли ми десь щось гасимо, кластер через 2-3 секунди рапортує, що в центрі А ми загасили ноду 2, 3, і 4 - це означає, що в інших дата-центрах ми ні в якому разі не можемо гасити ті ноди, на яких залишилися шарди в єдиному екземплярі.
- Знаючи характер поведінки майстра, ми дуже уважно дивимося на кількість pending-завдань. Тому що навіть одне завдання, що зависло, якщо вчасно не відтаймаутиться, теоретично в якійсь екстреній ситуації здатна стати тією причиною, з якої у нас не відпрацює, припустимо, промоушен replica-шарда в primary, через що постане індексація.
- Також ми дуже уважно дивимося на затримки garbage collector, тому що у нас із цим вже були великі складнощі при оптимізації.
- Реджекти за тредами, щоб розуміти заздалегідь, де знаходиться «пляшкове горло».
- Ну і стандартні метрики, типу heap, RAM та I/O.
При побудові моніторингу обов'язково треба враховувати особливості Thread Pool у Elasticsearch. описує можливості налаштування та дефолтні значення для пошуку, індексації, але повністю замовчує про thread_pool.management. Ці треди обробляють, зокрема, запити типу _cat/shards та інші аналогічні, які зручно використовувати під час написання моніторингу. Чим більше кластер, тим більше таких запитів виконується в одиницю часу, а вищезгаданий thread_pool.management мало того, що не представлений в офіційній документації, так ще й лімітований за дефолтом 5 тредами, що дуже швидко утилізується, після чого моніторинг перестає працювати коректно.
Що хочеться сказати на закінчення: у нас вийшло! Ми зуміли дати нашим програмістам та розробникам інструмент, який практично в будь-якій ситуації здатний швидко та достовірно надати інформацію про те, що відбувається на продакшені.
Так, це вийшло досить складно, але, проте, наші хотілки вдалося вкласти у вже існуючі продукти, які при цьому не довелося патчити і переписувати під себе.
Джерело: habr.com