ProHoster > Blog > Administração > Aplicação prática do ELK. Configurando o logstash

Aplicação prática do ELK. Configurando o logstash

Introdução

Ao implementar mais um sistema, nos deparamos com a necessidade de processar uma grande variedade de logs. Como ferramenta, escolhemos o ELK. Neste artigo, falaremos sobre nossa experiência na configuração desse stack.

Não temos a intenção de descrever todas as suas funcionalidades, mas queremos nos concentrar na solução de problemas práticos. Isso se deve ao fato de que, mesmo com uma quantidade considerável de documentação e imagens já prontas, existem muitos obstáculos, pelo menos encontramos isso.

Implantamos o stack através do docker-compose. Além disso, tínhamos um docker-compose.yml bem escrito, que nos permitiu levantar o stack praticamente sem problemas. Achávamos que a vitória já estava próxima, só precisávamos ajustar um pouco para nossas necessidades e seria tudo.

Infelizmente, a tentativa de ajustar o sistema para receber e processar logs do nosso aplicativo não teve sucesso de imediato. Portanto, decidimos que era melhor estudar cada componente separadamente e depois voltar para suas interconexões.

Então, começamos pelo logstash.

Ambiente, implantação, execução do Logstash em um contêiner

Para a implantação, estamos usando docker-compose; os experimentos descritos aqui foram realizados no MacOS e Ubuntu 18.0.4.

A imagem do logstash que temos no nosso docker-compose.yml é docker.elastic.co/logstash/logstash:6.3.2

Usaremos essa imagem para os experimentos.

Para executar o logstash, escrevemos um docker-compose.yml separado. Claro que poderíamos ter executado a imagem a partir da linha de comando, mas estávamos lidando com uma tarefa específica, onde tudo é iniciado a partir do docker-compose.

Breve sobre os arquivos de configuração

Como indicado na descrição, o logstash pode ser executado para um único canal, nesse caso, deve receber um arquivo *.conf, ou para vários canais, onde deve receber um arquivo pipelines.yml, que, por sua vez, fará referência aos arquivos .conf para cada canal.
Optamos pelo segundo caminho. Achamos que era mais universal e escalável. Portanto, criamos um pipelines.yml e montamos um diretório pipelines, onde colocaremos os arquivos .conf para cada canal.

Dentro do contêiner, há outro arquivo de configuração — logstash.yml. Não o alteramos, usamos como está.

Então, a estrutura de nossos diretórios:

Aplicação prática do ELK. Configurando o logstash

Para receber os dados de entrada, inicialmente consideramos que seria tcp na porta 5046, e para a saída usaremos stdout.

Aqui está uma configuração simples para a primeira execução. Afinal, o objetivo inicial é iniciar.

Então, temos o seguinte docker-compose.yml

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      	- elk
    ports:
      	- 5046:5046
    volumes:
      	- ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
	- ./config/pipelines:/usr/share/logstash/config/pipelines:ro

O que vemos aqui?

  1. As redes e volumes foram retirados do docker-compose.yml original (aquele onde todo o stack é iniciado) e acho que não impactam muito na visão geral.
  2. Estamos criando um serviço (services) logstash, a partir da imagem docker.elastic.co/logstash/logstash:6.3.2 e dando-lhe o nome logstash_one_channel.
  3. Estamos redirecionando a porta 5046 para dentro do contêiner, na mesma porta interna.
  4. Estamos mapeando nosso arquivo de configurações de canais ./config/pipelines.yml para o arquivo /usr/share/logstash/config/pipelines.yml dentro do contêiner, de onde será carregado pelo logstash, e o fazemos read-only, apenas por precaução.
  5. Estamos mapeando o diretório ./config/pipelines, onde temos os arquivos de configurações de canais, para o diretório /usr/share/logstash/config/pipelines e também o fazemos read-only.

Aplicação prática do ELK. Configurando o logstash

Arquivo pipelines.yml

- pipeline.id: HABR
  pipeline.workers: 1
  pipeline.batch.size: 1
  path.config: "./config/pipelines/habr_pipeline.conf"

Aqui está descrito um canal com o identificador HABR e o caminho para seu arquivo de configuração.

E finalmente o arquivo "./config/pipelines/habr_pipeline.conf"

input {
  tcp {
    port => "5046"
   }
  }
filter {
  mutate {
    add_field => [ "habra_field", "Hello Habr" ]
    }
  }
output {
  stdout {
      
    }
  }

Não vamos nos aprofundar em sua descrição por enquanto, vamos tentar iniciar:

docker-compose up

O que vemos?

O contêiner foi iniciado. Podemos verificar seu funcionamento:

echo '13123123123123123123123213123213' | nc localhost 5046

E vemos na consola do contêiner a resposta:

Aplicação prática do ELK. Configurando o logstash

Mas ao mesmo tempo, também vemos:

logstash_one_channel | [2019-04-29T11:28:59,790][ERROR][logstash.licensechecker.licensereader] Unable to retrieve license information from license server {:message=>"Elasticsearch Unreachable: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch", …

logstash_one_channel | [2019-04-29T11:28:59,894][INFO ][logstash.pipeline ] Pipeline started successfully {:pipeline_id=>".monitoring-logstash", :thread=>"#"}

logstash_one_channel | [2019-04-29T11:28:59,988][INFO ][logstash.agent ] Pipelines running {:count=>2, :running_pipelines=>[:HABR, ".monitoring-logstash"], :non_running_pipelines=>[]}
logstash_one_channel | [2019-04-29T11:29:00,015][ERROR][logstash.inputs.metrics ] X-Pack is installed on Logstash but not on Elasticsearch. Please install X-Pack on Elasticsearch to use the monitoring feature. Other features may be available.
logstash_one_channel | [2019-04-29T11:29:00,526][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600}
logstash_one_channel | [2019-04-29T11:29:04,478][INFO ][logstash.outputs.elasticsearch] Executando verificação de saúde para ver se uma conexão Elasticsearch está funcionando {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,487][WARN ][logstash.outputs.elasticsearch] Tentativa de ressuscitar conexão com instância ES inativa, mas ocorreu um erro. {:url=«elasticsearch:9200/, :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch Inacessível: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}
logstash_one_channel | [2019-04-29T11:29:04,704][INFO ][logstash.licensechecker.licensereader] Executando verificação de saúde para ver se uma conexão Elasticsearch está funcionando {:healthcheck_url=http://elasticsearch:9200/, :path="/"}
logstash_one_channel | [2019-04-29T11:29:04,710][WARN ][logstash.licensechecker.licensereader] Tentativa de ressuscitar conexão com instância ES inativa, mas ocorreu um erro. {:url=«elasticsearch:9200/, :error_type=LogStash::Outputs::ElasticSearch::HttpClient::Pool::HostUnreachableError, :error=«Elasticsearch Inacessível: [http://elasticsearch:9200/][Manticore::ResolutionFailure] elasticsearch»}

E nosso log está subindo constantemente.

Aqui destaquei em verde a mensagem de que o pipeline foi iniciado com sucesso, em vermelho — a mensagem de erro e em amarelo — a mensagem que tenta se conectar com elasticsearch:9200.
Isso ocorre porque no logstash.conf, incluído na imagem, há uma verificação de disponibilidade do elasticsearch. O logstash presume que está operando como parte da pilha Elk, e nós o separamos.

É possível trabalhar, mas não é conveniente.

A solução é desativar essa verificação através da variável de ambiente XPACK_MONITORING_ENABLED.

Vamos fazer a alteração no docker-compose.yml e reiniciar:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
    volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro

Agora sim, tudo certo. O container está pronto para experimentos.

Podemos novamente digitar no console ao lado:

echo '13123123123123123123123213123213' | nc localhost 5046

E ver:

logstash_one_channel | {
logstash_one_channel |         "message" => "13123123123123123123123213123213",
logstash_one_channel |      "@timestamp" => 2019-04-29T11:43:44.582Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |            "host" => "gateway",
logstash_one_channel |            "port" => 49418
logstash_one_channel | }

Trabalho em um único canal

Então, nós iniciamos. Agora podemos dedicar tempo à configuração do logstash. Não vamos mexer no arquivo pipelines.yml por enquanto, vamos ver o que podemos obter trabalhando com um único canal.

Devo dizer que o princípio geral de trabalho com o arquivo de configuração do canal está bem descrito no guia oficial, aqui está aqui
Se você quiser ler em português, usamos este artigo(mas a sintaxe das consultas está desatualizada, é preciso levar isso em conta).

Vamos seguir sequencialmente pela seção Input. Já vimos o trabalho com TCP. O que mais pode ser interessante aqui?

Mensagens de teste usando heartbeat

Há uma possibilidade interessante de gerar mensagens de teste automáticas.
Para isso, precisamos ativar o plugin heartbeat na seção input.

input {
  heartbeat {
    message => "HeartBeat!"
   }
  }

Ativamos e começamos a receber uma vez por minuto

logstash_one_channel | {
logstash_one_channel |      "@timestamp" => 2019-04-29T13:52:04.567Z,
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "HeartBeat!",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "host" => "a0667e5c57ec"
logstash_one_channel | }

Queremos receber com mais frequência, precisamos adicionar o parâmetro interval.
Assim, receberemos a mensagem a cada 10 segundos.

input {
  heartbeat {
    message => "HeartBeat!"
    interval => 10
   }
  }

Recebendo dados de um arquivo

Decidimos também testar o modo file. Se funcionar bem com o arquivo, talvez não precisemos de nenhum agente, pelo menos para uso local.

De acordo com a descrição, o modo de operação deve ser semelhante ao tail -f, ou seja, lê novas linhas ou, como opção, lê o arquivo inteiro.

Então, o que queremos obter:

  1. Queremos receber as linhas que são adicionadas a um arquivo de log.
  2. Queremos receber dados que são gravados em vários arquivos de log, tendo a capacidade de separar de onde cada um foi obtido.
  3. Queremos verificar se, ao reiniciar o logstash, ele não receberá esses dados novamente.
  4. Queremos verificar se, ao desligar o logstash, e os dados continuarem sendo gravados nos arquivos, quando ligarmos novamente, conseguiremos receber esses dados.

Para realizar o experimento, vamos adicionar mais uma linha ao docker-compose.yml, abrindo o diretório onde colocamos os arquivos.

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input

E vamos modificar a seção input no habr_pipeline.conf

input {
  file {
    path => "/usr/share/logstash/input/*.log"
   }
  }

Vamos iniciar:

docker-compose up

Para criar e gravar arquivos de log, usaremos o comando:

echo '1' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:53.876Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Sim, funciona!

Nesse caso, vemos que um campo de caminho foi adicionado automaticamente. Isso significa que poderemos filtrar as entradas com base nele daqui para frente.

Vamos tentar mais uma vez:

echo '2' >> logs/number1.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:28:59.906Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "2",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log"
logstash_one_channel | }

Agora em outro arquivo:

 echo '1' >> logs/number2.log


{
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:29:26.061Z,
logstash_one_channel |        "@version" => "1",
logstash_one_channel |         "message" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log"
logstash_one_channel | }

Ótimo! O arquivo foi capturado, o caminho foi definido corretamente, tudo está bom.

Vamos parar o logstash e reiniciar. Vamos esperar. Silêncio. Ou seja, não estamos recebendo essas entradas novamente.

E agora, o experimento mais ousado.

Vamos parar o logstash e executar:

echo '3' >> logs/number2.log
echo '4' >> logs/number1.log

Reiniciamos o logstash e vemos:

logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "3",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number2.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.589Z
logstash_one_channel | }
logstash_one_channel | {
logstash_one_channel |            "host" => "ac2d4e3ef70f",
logstash_one_channel |     "habra_field" => "Hello Habr",
logstash_one_channel |         "message" => "4",
logstash_one_channel |        "@version" => "1",
logstash_one_channel |            "path" => "/usr/share/logstash/input/number1.log",
logstash_one_channel |      "@timestamp" => 2019-04-29T14:48:50.856Z
logstash_one_channel | }

Uhul! Tudo foi capturado.

Mas, preciso alertar sobre o seguinte. Se o contêiner do logstash for removido (docker stop logstash_one_channel && docker rm logstash_one_channel), nada será capturado. A posição do arquivo, até onde foi lido, foi salva dentro do contêiner. Se você reiniciar 'do zero', ele aceitará apenas novas linhas.

Leitura de arquivos já existentes

Suponha que estamos iniciando o logstash pela primeira vez, mas já temos logs e gostaríamos de processá-los.
Se iniciarmos o logstash com a seção de input que usamos acima, não receberemos nada. Apenas novas linhas serão processadas pelo logstash.

Para que as linhas de arquivos existentes sejam puxadas, precisamos adicionar uma linha à seção input:

input {
  file {
    start_position => "beginning"
    path => "/usr/share/logstash/input/*.log"
   }
  }

Além disso, há um detalhe: isso só funciona para novos arquivos que o logstash ainda não viu. Para aqueles arquivos que já estiveram no radar do logstash, ele já memorizou seu tamanho e agora só pegará novas entradas.

Vamos parar por aqui no estudo da seção input. Existem muitas outras opções, mas, para nossos experimentos, por enquanto isso é suficiente.

Roteamento e transformação de dados

Vamos tentar resolver a seguinte tarefa: digamos que estamos recebendo mensagens de um canal, parte delas sendo informativas e parte mensagens de erro. Elas se diferenciam pela tag. Algumas são INFO, outras são ERROR.

Precisamos dividi-las na saída. Ou seja, as mensagens informativas devem ser escritas em um canal, enquanto as mensagens de erro devem ir para outro.

Para isso, vamos da seção input para filter e output.

Com a seção filter, analisaremos a mensagem de entrada, extraindo dela um hash (pares chave-valor), que já podemos usar para filtrar por condições. Na seção output, separaremos as mensagens e enviaremos cada uma para seu respectivo canal.

Análise de mensagens com grok

Para analisar strings de texto e extrair delas um conjunto de campos, na seção filter existe um plugin especial — grok.

Sem a intenção de fornecer uma descrição detalhada aqui (para isso, remeto a documentação oficial), apresentarei um exemplo simples.

Para isso, precisamos definir o formato das strings de entrada. Aqui estão elas:

1 INFO mensagem1
2 ERROR mensagem2

Ou seja, o identificador na primeira posição, depois INFO/ERROR, seguido de uma palavra sem espaços.
Não é difícil, mas é o suficiente para entender o princípio de funcionamento.

Assim, na seção filter, no plugin grok, devemos definir um padrão para analisar nossas strings.

Ele se parecerá com isto:

filter {
  grok {
    match => { "message" => ["%{INT:message_id} %{LOGLEVEL:message_type} %{WORD:message_text}"] }
   }
  }

Basicamente, isso é uma expressão regular. São utilizados padrões já prontos, como INT, LOGLEVEL, WORD. Sua descrição, assim como outros padrões, pode ser vista aqui aqui

Agora, passando por esse filtro, nossa string será convertida em um hash de três campos: message_id, message_type, message_text.

Esses serão os campos exibidos na seção output.

Roteamento de mensagens na seção output com o comando if

Na seção output, como lembramos, estávamos planejando dividir as mensagens em dois fluxos. Um — que é o INFO, vamos exibir no console, enquanto os erros, vamos registrar em um arquivo.

Como vamos dividir essas mensagens? A condição do problema já sugere a solução — já temos um campo destacado message_type, que pode assumir apenas dois valores: INFO e ERROR. Precisamos fazer a escolha com o operador if com base nisso.

if [message_type] == "ERROR" {
        # Aqui escrevemos no arquivo
       } else
     {
      # Aqui escrevemos no stdout
    }

A descrição do trabalho com campos e operadores pode ser vista nesta seção do manual oficial.

Agora, sobre a saída propriamente dita.

Saída no console, tudo claro — stdout {}

Já a saída em arquivo — lembremos que estamos executando tudo isso de um contêiner e, para que o arquivo em que escrevemos o resultado esteja acessível externamente, precisamos abrir esse diretório no docker-compose.yml.

No total:

A seção output do nosso arquivo fica assim:

output {
  if [message_type] == "ERROR" {
    file {
          path => "/usr/share/logstash/output/test.log"
          codec => line { format => "formato personalizado: %{message}"}
         }
    } else
     {stdout {
             }
     }
  }

No docker-compose.yml, adicionamos mais um volume, para a saída:

version: '3'

networks:
  elk:

volumes:
  elasticsearch:
    driver: local

services:

  logstash:
    container_name: logstash_one_channel
    image: docker.elastic.co/logstash/logstash:6.3.2
    networks:
      - elk
    environment:
      XPACK_MONITORING_ENABLED: "false"
    ports:
      - 5046:5046
   volumes:
      - ./config/pipelines.yml:/usr/share/logstash/config/pipelines.yml:ro
      - ./config/pipelines:/usr/share/logstash/config/pipelines:ro
      - ./logs:/usr/share/logstash/input
      - ./output:/usr/share/logstash/output

Iniciamos, testamos e vemos a divisão em dois fluxos.

Fonte: habr.com

Compre hospedagem confiável para sites com proteção contra DDoS, servidores VPS VDS 🔥 Compre hospedagem confiável para sites com proteção contra DDoS, servidores VPS VDS | ProHoster