Много хора се сблъскват с Elasticsearch. Но какво се случва, когато искаш да използваш него, за да съхраняваш логове "в особено голям обем"? И как да преживееш отказа на който и да е от многото дата центрове без последствия? Каква архитектура да изградим и на какви трудности може да се натъкнем?
Ние в Одноклассниках решихме да използваме Elasticsearch, за да решим проблема с лог мениджмънта, а сега споделяме опита си с Хабър: както архитектурата, така и трудностите.
Аз съм Пьотр Зацев, работя като системен администратор в Одноклассниках. Преди това също бях админ и работих с Manticore Search, Sphinx Search и Elasticsearch. Вероятно, ако се появи някакво ново решение за търсене, вероятно ще работя и с него. Участвам и в редица опенсорс проекти на доброволни начала.
Когато дойдох в Одноклассниках, глупаво казах на интервюто, че знам как да работя с 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. Има още два типа за различни преобразования на логове и свързване на различни клъстери помежду им, но ние използвахме само споменатите.
Master
Пингва всички налични в клъстера нодове, поддържа актуална карта на клъстера и разпространява нея между нодовете, обработва логиката на събитията, занимава се с различни дейности на клъстера.
Координатор
Изпълнява една единствена задача: приема заявки от клиенти за четене или запис и маршрутизира този трафик. В случай, че заявката е за запис, вероятно ще запита master в кой шард на релевантния индекс да я положи и ще пренасочи заявката по-нататък.
Data node
Съхранява данни, изпълнява поискови заявки и операции върху разположените на нея шарда.
Graylog
Това е нещо като комбиниране на 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 часа се търсят най-често).
Такъв интервал на ротация на индексите е свързан с следните причини:
Когато конкретната дата-нода получава търсене, по отношение на производителността е по-изгодно, когато се извиква един шард, ако размерът му е съпоставим с размера на хипа на нодата. Това позволява да се задържи 'горещата' част на индекса в хипа и да се достъпи бързо до нея. Когато 'горещите части' станат много, скоростта на търсене по индекса се влошава.
Когато нодата започне да извършва търсене на един шард, тя заделя брой нишки, равен на броя на хипертрединговите ядра на физическата машина. Ако търсенето обхваща голям брой шардове, броят на нишките нараства пропорционално. Това оказва негативно въздействие върху скоростта на търсене и отрицателно влияе на индексацията на нови данни.
За да осигурим необходимата латентност при търсене, решихме да използваме SSD. За бърза обработка на заявките машините, на които се разполагат тези контейнери, трябва да имат поне 56 ядра. Числото 56 е избрано като условно достатъчна стойност, определяща броя на нишките, които Elasticsearch ще генерира в процеса на работа. В Elasticsearch много параметри наThread pool зависят директно от броя на наличните ядра, което от своя страна пряко влияе на необходимия брой ноди в кластера по принципа "по-малко ядра — повече ноди."
В резултат на това получихме, че в средно един шард тежи около 20 гигабайта, и на 1 индекс се падат 360 шарда. Съответно, ако ги ротираме на всеки 48 часа, имаме 15 от тях. Всеки индекс съдържа данни за 2 дни.
Схемите за запис и четене на данни
Нека разберем как в тази система се записват данни.
Да предположим, че получаваме в координатора някаква заявка от Graylog. Например, искаме да индексираме 2-3 хиляди реда.
Координаторът, получавайки запрос от Graylog, допитва се до майстора: „В заявката за индексиране изрично укажахме индекс, но в който шард да се запише — не е посочено.“
Майсторът отговаря: „Запиши тази информация в шард номер 71“, след което тя се отправя директно към съответния дата-ноду, където се намира primary-shard номер 71.
След това логът на транзакциите се репликира на replica-shard, който вече се намира в друг дата-център.
От Graylog към координатора идва поисков запрос. Координаторът го пренасочва по индекса, като в същото време Elasticsearch на принципа round-robin разпределя заявките между primary-shard и replica-shard.
Нодовете с брой 180 отговарят неравномерно, и докато те отговарят, координаторът натрупва информация, която вече е била „изплюта“ от по-бързи дата-ноди. След това, когато или цялата информация е пристигнала, или по заявка е достигнат таймаут, всичко се предава директно на клиента.
Цялата тази система в средно време обработва поискови заявки по последните 48 часа за 300-400ms, с изключение на тези заявки, които имат leading wildcard.
„Цветята“ с Elasticsearch: настройка на Java
За да всичко това работи така, както първоначално желаехме, доста дълго настройвахме най-разнообразни неща в клъстера.
Първата част от откритите проблеми беше свързана с това, как Java по подразбиране е предварително настроена в Elasticsearch.
Проблема първа
Наблюдавахме много голямо количество съобщения относно това, че на ниво Lucene, когато се изпълняват фонова работа, сливането на сегменти Lucene завършва с грешка. В логовете беше видно, че това е OutOfMemoryError. По телеметрията виждахме, че хипът е свободен и не беше ясно защо тази операция пада.
Установи се, че сливането на Lucene индексите се извършва извън хипа. А контейнерите бяха доста строго ограничени по консумираните ресурси. В тези ресурси се побираше само хипът (стойността heap.size беше приблизително равна на RAM), а някакви off-heap операции паднаха с грешка на алокация на памет, ако по някаква причина не попаднаха в тези ~500MB, които останаха до лимита.
Фиксът беше доста тривиален: увеличихме наличния за контейнера обем RAM, след което забравихме, че такива проблеми изобщо сме имали.
Втори проблем
На около 4-5 ден след пускането на клъстера забелязахме, че дата-нодовете започват периодично да излизат от клъстера и да влизат в него след 10-20 секунди.
Когато започнахме да разглеждаме проблема, установихме, че паметта off-heap в Elasticsearch не се контролира по никакъв начин. Когато дадохме на контейнера повече памет, получихме възможност да запълваме директните пулове с различна информация, която се изчиствала само след като се активирал explicit GC от страна на Elasticsearch.
В някои случаи тази операция отнемаше доста време и през това време кластерът успяваше да маркира тази нода като вече излязла. Тази проблема е добре описана. .
Решението беше следното: ограничихме Java от възможността да използва основната част от паметта извън хипа за тези операции. Лимитирахме я до 16 гигабайта (-XX:MaxDirectMemorySize=16g), постигайки по-често извикване на explicit GC, който работеше значително по-бързо, като така не дестабилизираше кластера.
Проблемът трети.
Ако смятате, че проблемите с "нодите, напускащи клъстера в най-непредвидимия момент" свършиха тук, грешите.
Когато конфигурирахме работата с индексите, избрахме mmapfs, за да по свежи шардове с голяма сегментация. Това беше доста груба грешка, защото при използването на mmapfs файлът се мапва в оперативната памет, а след това работим вече с мапнатия файл. Заради това, когато GC се опитва да спре нишките в приложението, ние много бавно стигаме до safepoint и по пътя дотам приложението спира да отговаря на запитванията на мастера за това дали е живо или не. Според това, мастерът смята, че нода не е в клъстера. След около 5-10 секунди работи garbage collector, нода отново се съживява, влиза в клъстера и започва инициализация на шардовете. Всичко това важно наподобяваше “продакшен, който сме заслужили” и не беше подходящо за нищо сериозно.
За да се справим с такова поведение, първо преминахме на стандартния niofs, а след това, когато от версии пета на Elastic мигрирахме на шести, опитахме hybridfs, където тази проблема не се проявяваше. Повече информация за типове сторидж може да се прочете. .
Проблемът четвърти.
След това беше още една много интересна проблема, която лекувахме рекордно дълго. Ловяхме я 2-3 месеца, защото патернът ѝ абсолютно не ни беше ясен.
Понякога нашите координатори изпадаха в Full GC, обикновено следобед, и оттам не се завръщаха. При регистриране на забавяния на GC това изглеждаше така: всичко вървеше добре, добре, добре, а после изведнъж — всичко стана драстично лошо.
Първо мислехме, че имаме злонамерен потребител, който стартира някаква заявка, която избутва координатора от работен режим. Дълго записвахме заявките, опитвайки се да разберем какво се случва.
В крайна сметка се оказа, че в момента, когато някой потребител стартира огромна заявка и тя попада на конкретен координатор на Elasticsearch, някои нодове отговарят по-дълго от останалите.
И докато координаторът чака отговор от всички нодове, той натрупва резултати, изпратени от вече отговорилите нодове. За GC това означава, че много бързо се променя моделът на използване на хипа. И този GC, който използвахме, не се справяше с тази задача.
Едно единствено решение, което намерихме, за да променим поведението на клъстера в такава ситуация, беше миграцията на JDK13 и използването на garbage collector Shenandoah. Това реши проблема и координаторите ни спряха да падат.
С това проблемите с Java приключиха и започнаха проблемите с пропускната способност.
«Ягодките» с Elasticsearch: пропускна способност
Проблемите с пропускната способност означават, че нашият клъстер работи стабилно, но по време на пикове на индексируемите документи и в момент на маневри производителността не е достатъчна.
Първият забелязан симптом: при някакви «взривове» на продукцията, когато внезапно се генерира много голямо количество логове, в Graylog често започва да се появява грешка при индексация es_rejected_execution.
Това се случваше, защото thread_pool.write.queue на една дата-нода по време на обработката на заявка за индексация и поставяне на информация в шард на диск по подразбиране може да кешира само 200 заявки. И в за този параметър се говори много малко. Указва се само максималният брой нишки и подразбиращият се размер.
Разбира се, започнахме да настройваме тази стойност и установихме, че конкретно в нашия сетъп добре се кешират до 300 заявки, а по-голямото значение води до това, че отново попадаме в Full GC.
Освен това, тъй като това са пакети от съобщения, които пристигат в рамките на едно запитване, беше необходимо да се конфигурира Graylog, така че да записва не често и с малки партиди, а с огромни партиди или на всеки 3 секунди, ако партидата все още не е пълна. В такъв случай информацията, която записваме в Elasticsearch, става достъпна не за две секунди, а за пет (което ни устройва), но количеството повторни опити, които трябва да направим, за да прехвърлим голяма партида информация, намалява.
Това е особено важно в моментите, когато нещо е паднало някъде и яростно съобщава за себе си, за да не получим напълно задръстен Elastic, а след известно време — нефункциониращи нодули Graylog поради запушените буфери.
Освен това, когато се случваха тези взривове на продукция, получавахме оплаквания от програмисти и тестери: в моментите, когато имат много нужда от тези логове, те се предоставят много бавно.
Започнахме да разследваме. От една страна, беше ясно, че както търсенето, така и запитванията за индексиране се обработват по същество на одни и същи физически машини, и рано или късно определени спада ще възникнат.
Но това можеше частично да бъде избегнато благодарение на това, че в шестите версии на Elasticsearch се появи алгоритъм, който позволява да се разпределят запитванията между релевантни дата-нодове не по случаен принцип round-robin (контейнерът, който се занимава с индексация и държи primary-shard, може да бъде много натоварен, там не би имало възможност да отговори бързо), а да се насочи това запитване към по-малко натоварен контейнер с replica-shard, който ще отговори значително по-бързо. С други думи, стигнахме до use_adaptive_replica_selection: true.
Картината на четене започва да изглежда така:
Преминаването към този алгоритъм позволи значително подобряване на времето за заявка в моментите, когато имаше голям поток от логове за запис.
Накрая основният проблем беше безболезненото излизане от дата-центъра.
Какво искахме от клъстера веднага след загуба на връзка с един дата-център:
- Ако текущият 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 секунди докладва, че в центъра A сме затворили ноди 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
