Muitas pessoas têm dificuldades com o Elasticsearch. Mas o que acontece quando você deseja usá-lo para armazenar logs “em um volume particularmente grande”? E também é fácil experimentar a falha de qualquer um dos vários data centers? Que tipo de arquitetura você deve fazer e em quais armadilhas você encontrará?
Nós da Odnoklassniki decidimos usar o elasticsearch para resolver o problema do gerenciamento de logs e agora compartilhamos nossa experiência com Habr: tanto sobre arquitetura quanto sobre armadilhas.
Sou Pyotr Zaitsev e trabalho como administrador de sistema na Odnoklassniki. Antes também fui administrador, trabalhei com Manticore Search, Sphinx search, Elasticsearch. Talvez, se aparecer outra ...pesquisa, provavelmente eu também trabalharei com ela. Também participo de vários projetos de código aberto de forma voluntária.
Quando vim para Odnoklassniki, disse imprudentemente na entrevista que poderia trabalhar com o Elasticsearch. Depois de pegar o jeito e concluir algumas tarefas simples, recebi a grande tarefa de reformar o sistema de gerenciamento de logs que existia naquela época.
Requisitos
Os requisitos do sistema foram formulados da seguinte forma:
- Graylog seria usado como frontend. Como a empresa já tinha experiência no uso deste produto, os programadores e testadores sabiam disso, era familiar e conveniente para eles.
- Volume de dados: em média 50-80 mil mensagens por segundo, mas se algo quebrar, o tráfego não será limitado por nada, pode ser de 2 a 3 milhões de linhas por segundo
- Tendo discutido com os clientes os requisitos para a velocidade de processamento de consultas de pesquisa, percebemos que o padrão típico de uso de tal sistema é este: as pessoas estão procurando logs de seus aplicativos nos últimos dois dias e não querem esperar mais do que um segundo para o resultado de uma consulta formulada.
- Os administradores insistiram que o sistema fosse facilmente escalável, se necessário, sem exigir que se aprofundassem no seu funcionamento.
- Portanto, a única tarefa de manutenção que esses sistemas exigem periodicamente é a troca de algum hardware.
- Além disso, Odnoklassniki tem uma excelente tradição técnica: qualquer serviço que lançarmos deve sobreviver a uma falha do data center (repentina, não planejada e absolutamente a qualquer momento).
O último requisito na implementação deste projeto foi o que nos custou mais, do qual falarei com mais detalhes.
quarta-feira
Trabalhamos em quatro data centers, enquanto os nós de dados do Elasticsearch só podem ser localizados em três (por vários motivos não técnicos).
Esses quatro data centers contêm aproximadamente 18 mil fontes de log diferentes – hardware, contêineres, máquinas virtuais.
Recurso importante: o cluster começa em contêineres não em máquinas físicas, mas em . Os contêineres têm garantia de 2 núcleos, semelhante ao 2.0Ghz v4, com possibilidade de reciclagem dos demais núcleos caso estejam ociosos.
Em outras palavras:
Topologia
Inicialmente vi a forma geral da solução da seguinte forma:
- 3-4 VIPs estão por trás do registro A do domínio Graylog, este é o endereço para o qual os logs são enviados.
- cada VIP é um balanceador LVS.
- Depois disso, os logs vão para a bateria Graylog, alguns dados estão no formato GELF, outros no formato syslog.
- Depois, tudo isso é gravado em grandes lotes para uma bateria de coordenadores do Elasticsearch.
- E eles, por sua vez, enviam solicitações de gravação e leitura aos nós de dados relevantes.
Vocabulário
Talvez nem todos entendam a terminologia em detalhes, então gostaria de me alongar um pouco sobre ela.
Elasticsearch possui vários tipos de nós - mestre, coordenador, nó de dados. Existem dois outros tipos para diferentes transformações de log e comunicação entre diferentes clusters, mas usamos apenas os listados.
Mestre
Ele executa ping em todos os nós presentes no cluster, mantém um mapa de cluster atualizado e o distribui entre os nós, processa a lógica de eventos e executa vários tipos de manutenção em todo o cluster.
Coordenador
Executa uma única tarefa: aceita solicitações de leitura ou gravação de clientes e roteia esse tráfego. Caso haja uma solicitação de gravação, provavelmente ele perguntará ao master em qual fragmento do índice relevante ele deve colocá-lo e redirecionará a solicitação ainda mais.
Nó de dados
Armazena dados, realiza consultas de pesquisa que chegam de fora e executa operações em fragmentos nele localizados.
graylog
Isso é algo como uma fusão do Kibana com o Logstash em uma pilha ELK. Graylog combina uma UI e um pipeline de processamento de log. Nos bastidores, o Graylog executa o Kafka e o Zookeeper, que fornecem conectividade ao Graylog como um cluster. Graylog pode armazenar logs em cache (Kafka) caso o Elasticsearch esteja indisponível e repetir solicitações de leitura e gravação malsucedidas, agrupar e marcar logs de acordo com regras especificadas. Assim como o Logstash, o Graylog possui funcionalidade para modificar linhas antes de gravá-las no Elasticsearch.
Além disso, o Graylog possui uma descoberta de serviço integrada que permite, com base em um nó Elasticsearch disponível, obter todo o mapa do cluster e filtrá-lo por uma tag específica, o que possibilita direcionar solicitações para contêineres específicos.
Visualmente é algo assim:
Esta é uma captura de tela de uma instância específica. Aqui construímos um histograma com base na consulta de pesquisa e exibimos as linhas relevantes.
Índices
Voltando à arquitetura do sistema, gostaria de me aprofundar mais em como construímos o modelo de índice para que tudo funcionasse corretamente.
No diagrama acima, este é o nível mais baixo: nós de dados do Elasticsearch.
Um índice é uma grande entidade virtual composta de fragmentos do Elasticsearch. Por si só, cada um dos fragmentos nada mais é do que um índice Lucene. E cada índice Lucene, por sua vez, é composto por um ou mais segmentos.
Ao projetar, percebemos que, para atender ao requisito de velocidade de leitura em uma grande quantidade de dados, precisávamos “distribuir” esses dados uniformemente pelos nós de dados.
Isso resultou no fato de que o número de fragmentos por índice (com réplicas) deveria ser estritamente igual ao número de nós de dados. Em primeiro lugar, para garantir um fator de replicação igual a dois (ou seja, podemos perder metade do cluster). E, em segundo lugar, para processar solicitações de leitura e gravação em pelo menos metade do cluster.
Primeiro determinamos o tempo de armazenamento como 30 dias.
A distribuição dos fragmentos pode ser representada graficamente da seguinte forma:
Todo o retângulo cinza escuro é um índice. O quadrado vermelho esquerdo é o fragmento primário, o primeiro no índice. E o quadrado azul é um fragmento de réplica. Eles estão localizados em data centers diferentes.
Quando adicionamos outro fragmento, ele vai para o terceiro data center. E, no final, obtemos essa estrutura, que permite perder o DC sem perder a consistência dos dados:
Rotação de índices, ou seja, criando um novo índice e excluindo o mais antigo, igualamos 48 horas (de acordo com o padrão de uso do índice: as últimas 48 horas são pesquisadas com mais frequência).
Este intervalo de rotação do índice se deve aos seguintes motivos:
Quando uma solicitação de pesquisa chega a um nó de dados específico, então, do ponto de vista do desempenho, é mais lucrativo quando um fragmento é consultado, se seu tamanho for comparável ao tamanho do quadril do nó. Isso permite que você mantenha a parte “quente” do índice em uma pilha e acesse-a rapidamente. Quando há muitas “partes quentes”, a velocidade da pesquisa de índice diminui.
Quando um nó começa a executar uma consulta de pesquisa em um fragmento, ele aloca um número de threads igual ao número de núcleos de hyperthreading da máquina física. Se uma consulta de pesquisa afetar um grande número de fragmentos, o número de threads aumentará proporcionalmente. Isto tem um impacto negativo na velocidade de pesquisa e afeta negativamente a indexação de novos dados.
Para fornecer a latência de pesquisa necessária, decidimos usar um SSD. Para processar rapidamente as solicitações, as máquinas que hospedavam esses contêineres precisavam ter pelo menos 56 núcleos. O número 56 foi escolhido como um valor condicionalmente suficiente que determina o número de threads que o Elasticsearch irá gerar durante a operação. No Elaitcsearch, muitos parâmetros do pool de threads dependem diretamente do número de núcleos disponíveis, o que por sua vez afeta diretamente o número necessário de nós no cluster de acordo com o princípio “menos núcleos - mais nós”.
Como resultado, descobrimos que, em média, um fragmento pesa cerca de 20 gigabytes e existem 1 fragmentos por índice. Conseqüentemente, se os rotacionarmos uma vez a cada 360 horas, teremos 48 deles. Cada índice contém dados de 15 dias.
Circuitos de escrita e leitura de dados
Vamos descobrir como os dados são registrados neste sistema.
Digamos que chegue algum pedido do Graylog ao coordenador. Por exemplo, queremos indexar de 2 a 3 mil linhas.
O coordenador, ao receber uma solicitação do Graylog, questiona o mestre: “Na solicitação de indexação especificamos especificamente um índice, mas não foi especificado em qual shard escrever ele”.
O mestre responde: “Escreva esta informação no fragmento número 71”, após o que ela é enviada diretamente para o nó de dados relevante, onde o fragmento primário número 71 está localizado.
Depois disso, o log de transações é replicado para um fragmento de réplica localizado em outro data center.
Uma solicitação de busca chega do Graylog ao coordenador. O coordenador o redireciona de acordo com o índice, enquanto o Elasticsearch distribui as solicitações entre o fragmento primário e o fragmento de réplica usando o princípio round-robin.
Os 180 nós respondem de forma desigual e, enquanto respondem, o coordenador acumula informações que já foram “cuspidas” por nós de dados mais rápidos. Depois disso, quando todas as informações chegam ou a solicitação atinge o tempo limite, ele entrega tudo diretamente ao cliente.
Todo esse sistema processa, em média, consultas de pesquisa das últimas 48 horas em 300-400 ms, excluindo as consultas com um curinga inicial.
Flores com Elasticsearch: configuração Java
Para fazer tudo funcionar da maneira que queríamos originalmente, passamos muito tempo depurando uma grande variedade de coisas no cluster.
A primeira parte dos problemas descobertos estava relacionada à forma como o Java é pré-configurado por padrão no Elasticsearch.
O primeiro problema
Observamos um grande número de relatórios de que, no nível do Lucene, quando as tarefas em segundo plano estão em execução, as mesclagens de segmentos do Lucene falham com um erro. Ao mesmo tempo, ficou claro nos logs que se tratava de um erro OutOfMemoryError. Vimos pela telemetria que o quadril estava livre e não ficou claro por que essa operação estava falhando.
Descobriu-se que as mesclagens do índice Lucene ocorrem fora do quadril. E os contêineres são estritamente limitados em termos de recursos consumidos. Apenas o heap poderia caber nesses recursos (o valor heap.size era aproximadamente igual à RAM), e algumas operações fora do heap travavam com um erro de alocação de memória se, por algum motivo, não coubessem nos ~500 MB que restavam antes do limite.
A correção foi bastante trivial: a quantidade de RAM disponível para o contêiner foi aumentada, depois disso esquecemos que tínhamos esses problemas.
Problema dois
4 a 5 dias após o lançamento do cluster, notamos que os nós de dados começaram a cair periodicamente do cluster e a entrar nele após 10 a 20 segundos.
Quando começamos a descobrir isso, descobrimos que essa memória fora do heap no Elasticsearch não é controlada de forma alguma. Quando fornecemos mais memória ao contêiner, conseguimos preencher os buffer pools diretos com diversas informações, e isso foi limpo somente depois que o GC explícito foi iniciado no Elasticsearch.
Em alguns casos, esta operação demorava bastante e durante esse tempo o cluster conseguiu marcar este nó como já encerrado. Este problema está bem descrito .
A solução foi a seguinte: limitamos a capacidade do Java de usar a maior parte da memória fora do heap para essas operações. Nós o limitamos a 16 gigabytes (-XX:MaxDirectMemorySize=16g), garantindo que o GC explícito fosse chamado com muito mais frequência e processado com muito mais rapidez, não desestabilizando mais o cluster.
Problema três
Se você pensa que os problemas de “nós saindo do cluster no momento mais inesperado” acabaram, você está enganado.
Quando configuramos o trabalho com índices, escolhemos mmapfs para em fragmentos novos com ótima segmentação. Isso foi um grande erro, pois ao usar o mmapfs o arquivo é mapeado para a RAM, e então trabalhamos com o arquivo mapeado. Por conta disso, acontece que quando o GC tenta interromper threads na aplicação, levamos muito tempo para o ponto seguro e, no caminho até ele, a aplicação para de responder às solicitações do mestre sobre se está vivo . Conseqüentemente, o mestre acredita que o nó não está mais presente no cluster. Depois disso, após 5 a 10 segundos, o coletor de lixo funciona, o nó ganha vida, entra no cluster novamente e começa a inicializar os fragmentos. Tudo parecia “a produção que merecíamos” e não era adequado para nada sério.
Para nos livrar desse comportamento, primeiro mudamos para o niofs padrão e depois, quando migramos da quinta versão do Elastic para a sexta, tentamos o hybridfs, onde esse problema não foi reproduzido. Você pode ler mais sobre os tipos de armazenamento .
Problema quatro
Depois houve outro problema muito interessante que tratamos em tempo recorde. Nós o pegamos por 2 a 3 meses porque seu padrão era absolutamente incompreensível.
Às vezes nossos coordenadores iam para o Full GC, geralmente depois do almoço, e nunca mais voltavam de lá. Ao mesmo tempo, ao registrar o atraso do GC, ficou assim: tudo está indo bem, bem, bem, e de repente tudo está indo muito mal.
A princípio pensamos que tínhamos um usuário malvado que estava lançando algum tipo de solicitação que tirava o coordenador do modo de trabalho. Registramos solicitações por muito tempo, tentando descobrir o que estava acontecendo.
Como resultado, descobriu-se que no momento em que um usuário inicia uma solicitação enorme e chega a um coordenador específico do Elasticsearch, alguns nós respondem por mais tempo do que outros.
E enquanto o coordenador aguarda uma resposta de todos os nós, ele acumula os resultados enviados dos nós que já responderam. Para o GC, isso significa que nossos padrões de uso de heap mudam muito rapidamente. E o GC que usamos não deu conta dessa tarefa.
A única solução que encontramos para alterar o comportamento do cluster nesta situação é a migração para JDK13 e uso do coletor de lixo Shenandoah. Isso resolveu o problema, nossos coordenadores pararam de cair.
Foi aqui que os problemas com Java terminaram e começaram os problemas de largura de banda.
"Berries" com Elasticsearch: rendimento
Problemas de rendimento fazem com que nosso cluster funcione de forma estável, mas em picos de número de documentos indexados e durante manobras, o desempenho é insuficiente.
O primeiro sintoma encontrado: durante algumas “explosões” na produção, quando um número muito grande de logs é gerado repentinamente, o erro de indexação es_rejected_execution começa a piscar frequentemente no Graylog.
Isso ocorreu porque thread_pool.write.queue em um nó de dados, até o momento em que o Elasticsearch é capaz de processar a solicitação de indexação e fazer upload das informações para o fragmento no disco, é capaz de armazenar em cache apenas 200 solicitações por padrão. E em Muito pouco se fala sobre esse parâmetro. Apenas o número máximo de threads e o tamanho padrão são indicados.
É claro que distorcemos esse valor e descobrimos o seguinte: especificamente, em nossa configuração, até 300 solicitações são armazenadas em cache muito bem, e um valor mais alto está repleto do fato de voarmos novamente para o Full GC.
Além disso, como se trata de lotes de mensagens que chegam dentro de uma solicitação, foi necessário ajustar o Graylog para que ele não escrevesse com frequência e em pequenos lotes, mas em lotes enormes ou uma vez a cada 3 segundos se o lote ainda não estivesse completo. Neste caso, verifica-se que a informação que escrevemos no Elasticsearch fica disponível não em dois segundos, mas em cinco (o que nos convém muito bem), mas o número de retrays que devem ser feitos para passar por um grande pilha de informações é reduzida.
Isso é especialmente importante nos momentos em que algo travou em algum lugar e relata furiosamente sobre isso, para não receber um Elastic completamente spam, e depois de algum tempo - nós Graylog que estão inoperantes devido a buffers entupidos.
Além disso, quando tivemos essas mesmas explosões em produção, recebemos reclamações de programadores e testadores: no momento em que eles realmente precisavam desses logs, eles foram entregues muito lentamente.
Eles começaram a descobrir. Por um lado, estava claro que tanto as consultas de pesquisa como as consultas de indexação eram processadas, essencialmente, nas mesmas máquinas físicas e, de uma forma ou de outra, haveria certos rebaixamentos.
Mas isso poderia ser parcialmente contornado devido ao fato de que nas sextas versões do Elasticsearch, apareceu um algoritmo que permite distribuir consultas entre nós de dados relevantes, não de acordo com o princípio do round-robin aleatório (o contêiner que faz a indexação e mantém o primário -shard pode estar muito ocupado, não haverá como responder rapidamente), mas encaminhar essa solicitação para um contêiner menos carregado com um fragmento de réplica, que responderá muito mais rápido. Em outras palavras, chegamos a use_adaptive_replica_selection: true.
A imagem de leitura começa a ficar assim:
A transição para este algoritmo permitiu melhorar significativamente o tempo de consulta naqueles momentos em que tínhamos um grande fluxo de logs para escrever.
Finalmente, o principal problema foi a remoção indolor do data center.
O que queríamos do cluster imediatamente após perder a conexão com um DC:
- Se tivermos um mestre atual no data center com falha, ele será selecionado novamente e movido como uma função para outro nó em outro DC.
- O mestre removerá rapidamente todos os nós inacessíveis do cluster.
- Com base nos demais, ele entenderá: no data center perdido tínhamos tais e tais fragmentos primários, ele promoverá rapidamente fragmentos de réplica complementares nos data centers restantes e continuaremos indexando os dados.
- Como resultado disso, a taxa de transferência de escrita e leitura do cluster irá degradar gradualmente, mas em geral tudo funcionará, embora lentamente, mas de forma estável.
No final das contas, queríamos algo assim:
E obtivemos o seguinte:
Como isso aconteceu?
Quando o data center caiu, nosso mestre se tornou o gargalo.
Por quê?
O fato é que o mestre possui um TaskBatcher, que é responsável por distribuir determinadas tarefas e eventos no cluster. Qualquer saída de nó, qualquer promoção de um fragmento de réplica para primário, qualquer tarefa para criar um fragmento em algum lugar - tudo isso vai primeiro para o TaskBatcher, onde é processado sequencialmente e em um thread.
No momento da retirada de um data center, descobriu-se que todos os nós de dados nos data centers sobreviventes consideraram seu dever informar ao mestre “perdemos tais e tais fragmentos e tais e tais nós de dados”.
Ao mesmo tempo, os nós de dados sobreviventes enviaram todas essas informações ao mestre atual e tentaram aguardar a confirmação de que ele as aceitou. Eles não esperaram por isso, pois o mestre recebia as tarefas mais rápido do que conseguia responder. Os nós expiraram as solicitações repetidas, e o mestre neste momento nem tentou respondê-las, mas ficou completamente absorto na tarefa de classificar as solicitações por prioridade.
Na forma de terminal, descobriu-se que os nós de dados enviaram spam ao mestre a ponto de ele entrar em GC completo. Depois disso, nossa função mestre mudou para o próximo nó, absolutamente a mesma coisa aconteceu com ele e, como resultado, o cluster entrou em colapso completamente.
Fizemos medições e, antes da versão 6.4.0, onde isso foi corrigido, bastava gerarmos simultaneamente apenas 10 nós de dados de 360 para desligar completamente o cluster.
Parecia algo assim:
Após a versão 6.4.0, onde esse bug terrível foi corrigido, os nós de dados pararam de matar o mestre. Mas isso não o tornou “mais inteligente”. Ou seja: quando produzimos 2, 3 ou 10 (qualquer número diferente de um) nós de dados, o mestre recebe uma primeira mensagem que diz que o nó A saiu e tenta informar o nó B, nó C sobre isso, nó D.
E no momento, isso só pode ser resolvido definindo um tempo limite para tentativas de contar a alguém sobre algo, igual a cerca de 20 a 30 segundos, e assim controlar a velocidade de saída do data center do cluster.
Em princípio, isso se enquadra nos requisitos que foram inicialmente apresentados ao produto final como parte do projeto, mas do ponto de vista da “ciência pura” isso é um bug. Que, aliás, foi corrigido com sucesso pelos desenvolvedores na versão 7.2.
Além disso, quando um determinado nó de dados saiu, descobriu-se que a divulgação de informações sobre sua saída era mais importante do que dizer a todo o cluster que havia tais e tais fragmentos primários nele (a fim de promover um fragmento de réplica em outro nó de dados). centro no primário, e informações podem ser escritas neles).
Portanto, quando tudo já acabou, os nós de dados liberados não são imediatamente marcados como obsoletos. Conseqüentemente, somos forçados a esperar até que todos os pings atinjam o tempo limite para os nós de dados liberados, e somente depois disso nosso cluster começa a nos dizer que ali, ali e ali precisamos continuar registrando informações. Você pode ler mais sobre isso .
Com isso, a operação de retirada de um data center hoje leva cerca de 5 minutos no horário de pico. Para um colosso tão grande e desajeitado, este é um resultado muito bom.
Como resultado, chegamos à seguinte decisão:
- Temos 360 nós de dados com discos de 700 gigabytes.
- 60 coordenadores para rotear o tráfego através desses mesmos nós de dados.
- 40 masters que deixamos como uma espécie de legado desde as versões anteriores a 6.4.0 - para sobreviver à retirada do data center, estávamos mentalmente preparados para perder diversas máquinas para ter a garantia de ter um quorum de masters mesmo em o pior cenário
- Qualquer tentativa de combinar funções em um contêiner resultou no fato de que, mais cedo ou mais tarde, o nó quebraria sob carga.
- Todo o cluster usa um heap.size de 31 gigabytes: todas as tentativas de reduzir o tamanho resultaram na eliminação de alguns nós em consultas de pesquisa pesadas com o curinga inicial ou na obtenção do disjuntor no próprio Elasticsearch.
- Além disso, para garantir o desempenho da busca, procuramos manter o menor número de objetos no cluster possível, de forma a processar o menor número possível de eventos no gargalo que obtivemos no mestre.
Finalmente sobre monitoramento
Para garantir que tudo isso funcione conforme planejado, monitoramos o seguinte:
- Cada nó de dados informa à nossa nuvem que ele existe e que existem tais e tais fragmentos nele. Quando extinguimos algo em algum lugar, o cluster informa após 2-3 segundos que no centro A extinguimos os nós 2, 3 e 4 - isso significa que em outros data centers não podemos, sob nenhuma circunstância, extinguir os nós nos quais há apenas um fragmento esquerda.
- Conhecendo a natureza do comportamento do mestre, observamos com muito cuidado o número de tarefas pendentes. Porque mesmo uma tarefa travada, se não atingir o tempo limite, teoricamente em alguma situação de emergência pode se tornar o motivo pelo qual, por exemplo, a promoção de um fragmento de réplica no primário não funciona, razão pela qual a indexação irá parar de funcionar.
- Também observamos com atenção os atrasos do coletor de lixo, pois já tivemos grandes dificuldades com isso durante a otimização.
- Rejeita por fio para entender antecipadamente onde está o gargalo.
- Bem, métricas padrão como heap, RAM e E/S.
Ao construir o monitoramento, você deve levar em consideração os recursos do Thread Pool no Elasticsearch. descreve opções de configuração e valores padrão para pesquisa e indexação, mas é completamente silencioso sobre thread_pool.management.Esses threads processam, em particular, consultas como _cat/shards e outras semelhantes, que são convenientes para usar ao escrever monitoramento. Quanto maior o cluster, mais tais solicitações são executadas por unidade de tempo, e o mencionado thread_pool.management não só não é apresentado na documentação oficial, mas também é limitado por padrão a 5 threads, que são descartados muito rapidamente, após qual monitoramento para de funcionar corretamente.
O que quero dizer para concluir: conseguimos! Conseguimos fornecer aos nossos programadores e desenvolvedores uma ferramenta que, em quase todas as situações, pode fornecer informações de forma rápida e confiável sobre o que está acontecendo na produção.
Sim, acabou sendo bastante complicado, mas, mesmo assim, conseguimos encaixar nossos desejos nos produtos existentes, que não tivemos que corrigir e reescrever por nós mesmos.
Fonte: habr.com