Elasticsearch-kluster 200 TB+

Många människor kämpar med Elasticsearch. Men vad händer när du vill använda den för att lagra loggar "i en särskilt stor volym"? Och är det också smärtfritt att uppleva fel på något av flera datacenter? Vilken typ av arkitektur ska du göra, och vilka fallgropar kommer du att stöta på?

Vi på Odnoklassniki bestämde oss för att använda elasticsearch för att lösa problemet med stockhantering, och nu delar vi vår erfarenhet med Habr: både om arkitektur och om fallgropar.

Jag är Pyotr Zaitsev, jag arbetar som systemadministratör på Odnoklassniki. Innan dess var jag även admin, arbetade med Manticore Search, Sphinx search, Elasticsearch. Kanske, om en annan ...sökning dyker upp, kommer jag förmodligen att arbeta med den också. Jag deltar även i ett antal open source-projekt på frivillig basis.

När jag kom till Odnoklassniki sa jag hänsynslöst vid intervjun att jag kunde arbeta med Elasticsearch. Efter att jag fått kläm på det och gjort några enkla uppgifter fick jag den stora uppgiften att reformera det logghanteringssystem som fanns på den tiden.

Krav

Systemkraven formulerades enligt följande:

• Graylog skulle användas som frontend. Eftersom företaget redan hade erfarenhet av att använda den här produkten visste programmerare och testare det, det var bekant och bekvämt för dem.
• Datavolym: i genomsnitt 50-80 tusen meddelanden per sekund, men om något går sönder, begränsas inte trafiken av någonting, det kan vara 2-3 miljoner rader per sekund
• Efter att ha diskuterat med kunderna kraven på hastigheten för att bearbeta sökfrågor, insåg vi att det typiska mönstret för att använda ett sådant system är detta: människor letar efter loggar för sin ansökan under de senaste två dagarna och vill inte vänta mer än en andra för resultatet av en formulerad fråga.
• Administratörerna insisterade på att systemet skulle vara lätt skalbart om det skulle behövas, utan att kräva att de fördjupade sig i hur det fungerar.
• Så att den enda underhållsuppgiften som dessa system kräver regelbundet är att byta hårdvara.
• Dessutom har Odnoklassniki en utmärkt teknisk tradition: alla tjänster som vi lanserar måste överleva ett datacenterfel (plötsligt, oplanerat och absolut när som helst).

Det sista kravet i genomförandet av detta projekt kostade oss mest, vilket jag kommer att prata om mer i detalj.

onsdag

Vi arbetar i fyra datacenter, medan Elasticsearch-datanoder endast kan placeras i tre (av ett antal icke-tekniska skäl).

Dessa fyra datacenter innehåller cirka 18 tusen olika loggkällor - hårdvara, behållare, virtuella maskiner.

Viktig funktion: klustret startar i behållare poddman inte på fysiska maskiner, utan på egen molnprodukt one-cloud. Behållare är garanterade 2 kärnor, liknande 2.0Ghz v4, med möjlighet att återvinna de återstående kärnorna om de är inaktiva.

Med andra ord:

Topologi

Jag såg först den allmänna formen av lösningen som följer:

• 3-4 VIP:s ligger bakom A-posten för Graylog-domänen, detta är adressen som loggarna skickas till.
• varje VIP är en LVS-balanserare.
• Efter det går loggarna till Graylog-batteriet, en del av datan är i GELF-format, en del i syslog-format.
• Sedan skrivs allt detta i stora omgångar till ett batteri av Elasticsearch-koordinatorer.
• Och de skickar i sin tur skriv- och läsbegäranden till de relevanta datanoderna.

terminologi

Kanske inte alla förstår terminologin i detalj, så jag skulle vilja uppehålla mig lite vid den.

Elasticsearch har flera typer av noder - master, koordinator, datanod. Det finns två andra typer för olika loggtransformationer och kommunikation mellan olika kluster, men vi använde bara de listade.

Mästare
Den pingar alla noder som finns i klustret, upprätthåller en uppdaterad klusterkarta och distribuerar den mellan noder, bearbetar händelselogik och utför olika typer av klusteromfattande hushållning.

Samordnare
Utför en enda uppgift: accepterar läs- eller skrivförfrågningar från klienter och dirigerar denna trafik. Om det finns en skrivbegäran kommer den troligtvis att fråga master vilken bit av det relevanta indexet den ska lägga in den i, och kommer att omdirigera begäran vidare.

Datanod
Lagrar data, utför sökfrågor som kommer utifrån och utför operationer på skärvor som finns på den.

grålogg
Det här är ungefär en fusion av Kibana med Logstash i en ELK-stack. Graylog kombinerar både ett användargränssnitt och en loggbearbetningspipeline. Under huven driver Graylog Kafka och Zookeeper, som ger anslutning till Graylog som ett kluster. Graylog kan cacheloggar (Kafka) om Elasticsearch inte är tillgängligt och upprepa misslyckade läs- och skrivförfrågningar, gruppera och markera loggar enligt specificerade regler. Precis som Logstash har Graylog funktionalitet för att modifiera rader innan du skriver dem till Elasticsearch.

Dessutom har Graylog en inbyggd tjänsteupptäckt som gör det möjligt att, baserat på en tillgänglig Elasticsearch-nod, hämta hela klusterkartan och filtrera den efter en specifik tagg, vilket gör det möjligt att rikta förfrågningar till specifika behållare.

Visuellt ser det ut ungefär så här:

Detta är en skärmdump från en specifik instans. Här bygger vi ett histogram baserat på sökfrågan och visar relevanta rader.

Index

För att återgå till systemarkitekturen skulle jag vilja uppehålla mig mer i detalj vid hur vi byggde upp indexmodellen så att det hela fungerade korrekt.

I diagrammet ovan är detta den lägsta nivån: Elasticsearch-datanoder.

Ett index är en stor virtuell enhet som består av Elasticsearch-skärvor. I sig är var och en av skärvorna inget annat än ett Lucene-index. Och varje Lucene-index består i sin tur av ett eller flera segment.

När vi designade tänkte vi att för att uppfylla kravet på läshastighet på en stor mängd data, behövde vi "sprida" denna data jämnt över datanoder.

Detta resulterade i att antalet shards per index (med repliker) bör vara strikt lika med antalet datanoder. För det första, för att säkerställa en replikeringsfaktor lika med två (det vill säga, vi kan förlora hälften av klustret). Och, för det andra, för att bearbeta läs- och skrivförfrågningar på minst hälften av klustret.

Vi bestämde först lagringstiden till 30 dagar.

Fördelningen av skärvor kan representeras grafiskt enligt följande:

Hela den mörkgrå rektangeln är ett index. Den vänstra röda fyrkanten i den är den primära skärvan, den första i indexet. Och den blå fyrkanten är en replikskärva. De finns i olika datacenter.

När vi lägger till ytterligare en shard går den till det tredje datacentret. Och i slutändan får vi den här strukturen, som gör det möjligt att förlora DC utan att förlora datakonsistens:

Rotation av index, d.v.s. skapa ett nytt index och ta bort det äldsta, gjorde vi det lika med 48 timmar (enligt mönstret för indexanvändning: de senaste 48 timmarna söks oftast).

Detta indexrotationsintervall beror på följande skäl:

När en sökbegäran anländer till en specifik datanod är det, ur prestandasynpunkt, mer lönsamt när en shard efterfrågas, om dess storlek är jämförbar med storleken på nodens höft. Detta gör att du kan hålla den "heta" delen av indexet i en hög och snabbt komma åt det. När det finns många "heta delar" försämras hastigheten på indexsökningen.

När en nod börjar köra en sökfråga på en shard, allokerar den ett antal trådar lika med antalet hyperthreading-kärnor i den fysiska maskinen. Om en sökfråga påverkar ett stort antal skärvor, ökar antalet trådar proportionellt. Detta har en negativ inverkan på sökhastigheten och påverkar indexeringen av ny data negativt.

För att tillhandahålla den nödvändiga sökfördröjningen bestämde vi oss för att använda en SSD. För att snabbt kunna bearbeta förfrågningar måste maskinerna som var värd för dessa behållare ha minst 56 kärnor. Siffran 56 valdes som ett villkorligt tillräckligt värde som bestämmer antalet trådar som Elasticsearch kommer att generera under drift. I Elasitcsearch är många trådpoolsparametrar direkt beroende av antalet tillgängliga kärnor, vilket i sin tur direkt påverkar det nödvändiga antalet noder i klustret enligt principen "färre kärnor - fler noder".

Som ett resultat fann vi att en skärva i genomsnitt väger cirka 20 gigabyte, och det finns 1 skärvor per index. Följaktligen, om vi roterar dem en gång var 360:e timme, så har vi 48 av dem. Varje index innehåller data för 15 dagar.

Dataskrivande och läskretsar

Låt oss ta reda på hur data registreras i det här systemet.

Låt oss säga att någon förfrågan kommer från Graylog till koordinatorn. Till exempel vill vi indexera 2-3 tusen rader.

Samordnaren, efter att ha fått en förfrågan från Graylog, ifrågasätter befälhavaren: "I indexeringsbegäran angav vi specifikt ett index, men i vilket fragment att skriva det specificerades inte."

Mastern svarar: "Skriv denna information till shard nummer 71," varefter den skickas direkt till den relevanta datanoden, där primär shard nummer 71 finns.

Därefter replikeras transaktionsloggen till en replika-shard, som finns i ett annat datacenter.

En sökförfrågan kommer från Graylog till koordinatorn. Koordinatorn omdirigerar den enligt indexet, medan Elasticsearch distribuerar förfrågningar mellan primär-shard och replika-shard med hjälp av round-robin-principen.

De 180 noderna svarar ojämnt, och medan de svarar samlar koordinatorn på information som redan har "spottats ut" av snabbare datanoder. Efter detta, när antingen all information har kommit, eller begäran har nått en timeout, ger den allt direkt till klienten.

Hela det här systemet bearbetar i genomsnitt sökfrågor under de senaste 48 timmarna i 300-400 ms, exklusive de frågor med ett ledande jokertecken.

Blommor med Elasticsearch: Java-installation

För att få det hela att fungera som vi ursprungligen ville, spenderade vi mycket lång tid på att felsöka en mängd olika saker i klustret.

Den första delen av problemen som upptäcktes var relaterad till hur Java är förkonfigurerat som standard i Elasticsearch.

Problem ett
Vi har observerat ett mycket stort antal rapporter om att på Lucene-nivå, när bakgrundsjobb körs, misslyckas Lucene-segmentsammanslagningar med ett fel. Samtidigt stod det tydligt i loggarna att detta var ett OutOfMemoryError-fel. Vi såg från telemetri att höften var fri, och det var inte klart varför denna operation misslyckades.

Det visade sig att Lucene index sammanslagningar förekommer utanför höften. Och containrar är ganska strikt begränsade när det gäller förbrukade resurser. Endast heap kunde passa in i dessa resurser (heap.size-värdet var ungefär lika med RAM), och vissa off-heap-operationer kraschade med ett minnesallokeringsfel om de av någon anledning inte passade in i ~500MB som återstod innan gränsen.

Fixningen var ganska trivial: mängden RAM tillgängligt för behållaren ökades, varefter vi glömde att vi till och med hade sådana problem.

Problem två
4-5 dagar efter lanseringen av klustret märkte vi att datanoder periodvis började falla ut ur klustret och komma in i det efter 10-20 sekunder.

När vi började lista ut det visade det sig att detta off-heap-minne i Elasticsearch inte styrs på något sätt. När vi gav mer minne till behållaren kunde vi fylla de direkta buffertpoolerna med olika information, och den rensades först efter att den explicita GC:en lanserades från Elasticsearch.

I vissa fall tog denna operation ganska lång tid, och under denna tid lyckades klustret markera denna nod som redan avslutad. Detta problem är väl beskrivet här.

Lösningen var följande: vi begränsade Javas förmåga att använda huvuddelen av minnet utanför högen för dessa operationer. Vi begränsade den till 16 gigabyte (-XX:MaxDirectMemorySize=16g), vilket säkerställde att explicit GC anropades mycket oftare och bearbetades mycket snabbare, vilket inte längre destabiliserade klustret.

Problem tre
Om du tror att problemen med "noder som lämnar klustret i det mest oväntade ögonblicket" är över, har du fel.

När vi konfigurerade arbetet med index valde vi mmapfs till minska söktiden på färska skärvor med stor segmentering. Detta var en hel misstag, för när man använder mmapfs mappas filen till RAM, och sedan arbetar vi med den mappade filen. På grund av detta visar det sig att när GC försöker stoppa trådar i applikationen, går vi till safepointen under mycket lång tid, och på vägen dit slutar applikationen att svara på mästarens förfrågningar om huruvida den är vid liv . Följaktligen anser master att noden inte längre finns i klustret. Efter detta, efter 5-10 sekunder, fungerar sopsamlaren, noden kommer till liv, går in i klustret igen och börjar initialisera skärvor. Det hela kändes väldigt mycket som "den produktion vi förtjänade" och lämpade sig inte för något allvarligt.

För att bli av med detta beteende bytte vi först till standard niofs, och sedan, när vi migrerade från den femte versionen av Elastic till den sjätte, provade vi hybridfs, där detta problem inte reproducerades. Du kan läsa mer om lagringstyper här.

Problem fyra
Sedan var det ett annat mycket intressant problem som vi behandlade rekordtid. Vi fångade den i 2-3 månader eftersom dess mönster var helt obegripligt.

Ibland gick våra koordinatorer till Full GC, vanligtvis någon gång efter lunch, och kom aldrig tillbaka därifrån. Samtidigt, när man loggar GC-fördröjningen, såg det ut så här: allt går bra, bra, bra, och så plötsligt går allt väldigt dåligt.

Först trodde vi att vi hade en ond användare som lanserade någon form av begäran som slog koordinatorn ur arbetsläge. Vi loggade förfrågningar under mycket lång tid för att försöka ta reda på vad som hände.

Som ett resultat visade det sig att i det ögonblick när en användare lanserar en enorm förfrågan och den kommer till en specifik Elasticsearch-koordinator, svarar vissa noder längre än andra.

Och medan samordnaren väntar på ett svar från alla noder, samlar han på sig resultaten som skickas från de noder som redan har svarat. För GC betyder detta att våra heapanvändningsmönster förändras mycket snabbt. Och GC som vi använde kunde inte klara av denna uppgift.

Den enda lösningen vi hittade för att ändra beteendet hos klustret i den här situationen är migrering till JDK13 och användning av Shenandoah-sopsamlaren. Detta löste problemet, våra koordinatorer slutade falla.

Det var här problemen med Java slutade och bandbreddsproblemen började.

"Bär" med Elasticsearch: genomströmning

Problem med genomströmning gör att vårt kluster fungerar stabilt, men vid toppar i antalet indexerade dokument och under manövrar är prestandan otillräcklig.

Det första symtomet som påträffades: under vissa "explosioner" i produktionen, när ett mycket stort antal loggar plötsligt genereras, börjar indexeringsfelet es_rejected_execution att blinka ofta i Graylog.

Detta berodde på det faktum att thread_pool.write.queue på en datanod, tills det ögonblick Elasticsearch kan bearbeta indexeringsförfrågan och ladda upp informationen till fragmentet på disken, kan cachelagra endast 200 förfrågningar som standard. Och i Elasticsearch dokumentation Mycket lite sägs om denna parameter. Endast det maximala antalet trådar och standardstorleken anges.

Naturligtvis gick vi för att vrida på det här värdet och fick reda på följande: specifikt, i vår installation, cachelagras upp till 300 förfrågningar ganska bra, och ett högre värde är fyllt av det faktum att vi återigen flyger till Full GC.

Dessutom, eftersom det här är satser av meddelanden som kommer inom en begäran, var det nödvändigt att justera Graylog så att den inte skriver ofta och i små satser, utan i stora satser eller en gång var tredje sekund om partiet fortfarande inte är komplett. I det här fallet visar det sig att informationen som vi skriver i Elasticsearch blir tillgänglig inte på två sekunder, utan på fem (vilket passar oss ganska bra), men antalet återupptagningar som måste göras för att driva igenom en stor mängden information minskar.

Detta är särskilt viktigt i de ögonblick då något har kraschat någonstans och rasande rapporterar om det, för att inte få en helt spammad Elastic, och efter en tid - Graylog-noder som inte fungerar på grund av igensatta buffertar.

Dessutom, när vi hade samma explosioner i produktionen, fick vi klagomål från programmerare och testare: i det ögonblick då de verkligen behövde dessa loggar, fick de dem mycket långsamt.

De började komma på det. Å ena sidan var det tydligt att både sökfrågor och indexeringsfrågor bearbetades, i huvudsak på samma fysiska maskiner, och på ett eller annat sätt skulle det finnas vissa neddragningar.

Men detta kan delvis kringgås på grund av det faktum att det i den sjätte versionen av Elasticsearch dök upp en algoritm som låter dig distribuera frågor mellan relevanta datanoder som inte följer den slumpmässiga round-robin-principen (behållaren som gör indexering och håller den primära -shard kan vara mycket upptagen, det kommer inte att finnas något sätt att svara snabbt), utan att vidarebefordra denna begäran till en mindre laddad container med en replika-shard, som kommer att svara mycket snabbare. Med andra ord kom vi fram till use_adaptive_replica_selection: true.

Läsbilden börjar se ut så här:

Övergången till denna algoritm gjorde det möjligt att avsevärt förbättra frågetiden i de ögonblick då vi hade ett stort flöde av loggar att skriva.

Slutligen var huvudproblemet det smärtfria avlägsnandet av datacentret.

Vad vi ville ha från klustret omedelbart efter att vi förlorat anslutningen till en DC:

• Om vi ​​har en aktuell master i det misslyckade datacentret, kommer den att väljas om och flyttas som en roll till en annan nod i en annan DC.
• Mastern tar snabbt bort alla otillgängliga noder från klustret.
• Baserat på de återstående kommer han att förstå: i det förlorade datacentret hade vi sådana och sådana primära skärvor, han kommer snabbt att marknadsföra kompletterande replikskärvor i de återstående datacentren, och vi kommer att fortsätta indexera data.
• Som ett resultat av detta kommer klustrets skriv- och läskapacitet gradvis att försämras, men i allmänhet kommer allt att fungera, om än långsamt, men stabilt.

Som det visade sig ville vi ha något sånt här:

Och vi fick följande:

Hur hände det?

När datacentret föll blev vår mästare flaskhalsen.

Varför?

Faktum är att mastern har en TaskBatcher, som ansvarar för att distribuera vissa uppgifter och händelser i klustret. Alla nodutgångar, all marknadsföring av en shard från replik till primär, vilken uppgift som helst att skapa en shard någonstans - allt detta går först till TaskBatcher, där det bearbetas sekventiellt och i en tråd.

Vid tidpunkten för tillbakadragandet av ett datacenter visade det sig att alla datanoder i de överlevande datacenterna ansåg det vara sin plikt att informera befälhavaren "vi har tappat sådana och sådana skärvor och sådana och sådana datanoder."

Samtidigt skickade de överlevande datanoderna all denna information till den nuvarande mastern och försökte vänta på bekräftelse på att han accepterade den. De väntade inte på detta, eftersom befälhavaren fick uppgifter snabbare än han kunde svara på. Noderna tog timeout för upprepade förfrågningar, och befälhavaren försökte vid denna tidpunkt inte ens svara på dem, utan var helt upptagen i uppgiften att sortera förfrågningar efter prioritet.

I terminalform visade det sig att datanoderna spammade mastern till den grad att den gick in i full GC. Efter det flyttade vår mästarroll till någon nästa nod, absolut samma sak hände med den, och som ett resultat kollapsade klustret helt.

Vi gjorde mätningar, och innan version 6.4.0, där detta fixades, räckte det för oss att samtidigt bara mata ut 10 datanoder av 360 för att helt stänga av klustret.

Det såg ut ungefär så här:

Efter version 6.4.0, där denna fruktansvärda bugg fixades, slutade datanoder att döda mastern. Men det gjorde honom inte "smartare". Nämligen: när vi matar ut 2, 3 eller 10 (vilket som helst annat än en) datanoder, får mastern ett första meddelande som säger att nod A har lämnat, och försöker berätta för nod B, nod C om detta, nod D.

Och för tillfället kan detta bara hanteras genom att sätta en timeout för försök att berätta för någon om något, lika med cirka 20-30 sekunder, och därmed styra hastigheten på datacentret som flyttar ut ur klustret.

I princip passar detta in i de krav som från början presenterades på slutprodukten som en del av projektet, men ur "ren vetenskap" synvinkel är detta en bugg. Vilket för övrigt framgångsrikt fixades av utvecklarna i version 7.2.

Dessutom, när en viss datanod gick ut, visade det sig att spridning av information om dess utgång var viktigare än att berätta för hela klustret att det fanns sådana och sådana primära skärvor på den (för att främja en replika-shard i en annan data centrum i den primära, och i information kan skrivas på dem).

Därför, när allt redan har dött ut, markeras inte de släppta datanoderna omedelbart som inaktuella. Följaktligen tvingas vi vänta tills alla pingar har gått ut till de släppta datanoderna, och först efter det börjar vårt kluster berätta att där, där och där måste vi fortsätta att registrera information. Du kan läsa mer om detta här.

Som ett resultat av detta tar operationen att dra in ett datacenter idag cirka 5 minuter under rusningstid. För en så stor och klumpig koloss är detta ett ganska bra resultat.

Som ett resultat kom vi till följande beslut:

• Vi har 360 datanoder med 700 gigabyte diskar.
• 60 koordinatorer för att dirigera trafik genom samma datanoder.
• 40 masters som vi har lämnat som ett slags arv sedan versioner före 6.4.0 - för att överleva indragningen av datacentret var vi mentalt beredda att förlora flera maskiner för att garanteras ha ett kvorum av masters även i det värsta scenariot
• Alla försök att kombinera roller på en container möttes av det faktum att noden förr eller senare skulle gå sönder under belastning.
• Hela klustret använder en heap.size på 31 gigabyte: alla försök att minska storleken resulterade i att antingen döda några noder på tunga sökfrågor med det ledande jokertecknet eller att få strömbrytaren i Elasticsearch själv.
• Dessutom, för att säkerställa sökprestanda, försökte vi hålla antalet objekt i klustret så litet som möjligt, för att bearbeta så få händelser som möjligt i flaskhalsen som vi fick i mastern.

Äntligen om övervakning

För att säkerställa att allt detta fungerar som avsett övervakar vi följande:

• Varje datanod rapporterar till vårt moln att den finns, och att det finns sådana och sådana skärvor på den. När vi släcker något någonstans rapporterar klustret efter 2-3 sekunder att vi i centrum A släckte noderna 2, 3 och 4 - det betyder att vi i andra datacenter under inga omständigheter kan släcka de noder där det bara finns en skärva vänster.
• Genom att känna till arten av mästarens beteende tittar vi mycket noggrant på antalet pågående uppgifter. Eftersom även en uppgift som har fastnat, om den inte tar time-out i tid, teoretiskt i någon nödsituation kan bli orsaken till att till exempel främjandet av en replikskärva i den primära inte fungerar, vilket är anledningen till att indexeringen slutar fungera.
• Vi tittar också mycket noga på förseningar av sophämtare, eftersom vi redan har haft stora svårigheter med detta under optimeringen.
• Avvisar per tråd för att i förväg förstå var flaskhalsen finns.
• Tja, standardmått som heap, RAM och I/O.

När du bygger övervakning måste du ta hänsyn till funktionerna i Thread Pool i Elasticsearch. Elasticsearch dokumentation beskriver konfigurationsalternativ och standardvärden för sökning och indexering, men är helt tyst om thread_pool.management. Dessa trådar behandlar i synnerhet frågor som _cat/shards och andra liknande, som är bekväma att använda när man skriver övervakning. Ju större klustret är, desto fler sådana förfrågningar exekveras per tidsenhet, och den tidigare nämnda thread_pool.management presenteras inte bara inte i den officiella dokumentationen, utan är också som standard begränsad till 5 trådar, som mycket snabbt kasseras, efter vilken övervakning slutar fungera korrekt.

Vad jag vill säga avslutningsvis: vi gjorde det! Vi kunde ge våra programmerare och utvecklare ett verktyg som i nästan alla situationer snabbt och tillförlitligt kan ge information om vad som händer i produktionen.

Ja, det visade sig vara ganska komplicerat, men vi lyckades ändå passa in våra önskemål i befintliga produkter, som vi inte behövde lappa och skriva om för oss själva.

Källa: will.com