Dec 4th 2022: [PT] Coletando JSON logs com Elastic-Agent (e/ou Filebeat)

[TiagoQueiroz]

Este artigo também está disponível em inglês.

Logs estruturados particamente se tornaram um padrão na industria de software, permitindo um fácil entendimento e processamento dos dados. O Elastic-Agent oferece um grande número de integrações que permitem a fácil ingestão e processamento de log, entretanto nossas applicações não seguem nenhum padrão, ou seguem um padrão próprio, nesse caso precismos de algumas configurações especiais.

Elastic-Agent x Filebeat

• Elastic-Agent é uma forma única e unificada de adicionar monitoramento para logs, métricas e outros tipos de dados a um host.
• Filebeat é um expedidor leve para o envio e centralização de logs.

Qual deles devo usar? O Elastic-Agent proporciona a melhor experiência, tudo pode ser configurado via Kibana, você pode ver os logs do Elastic-Agent no Kibana, bem como o seu status de 'saúde'.

Se por algum motivo você não puder usar o Elastic-Agent, Filebeat ainda é uma opção, nós cobriremos sua configuração no final do artigo.

0. O básico do Elastic-Agent

Este artigo assume que você está familiarizado com o Elastic-Agent, já tem um rodando e entende conceitos como integrações, 'policy', etc.

1. Log em JSON

Simples assim. Se seus logs são objetos JSON, o Elastic-Agent já pode processa-los, então se torna uma questão de ajuste fino para garantir que o 'timestamp' seja ingerido corretamente e que os campos tenham o tipo correto.

2. Exemplo de logs

Vamos usar esses logs como exemplo:

{"level":"info","time":"2022-11-28T12:00:00+01:00","message":"Starting Advent Calendar demo", "status_code": 200}
{"level":"info","time":"2022-11-28T12:00:32+01:00","message":"First line", "status_code": 300}
{"level":"debug","time":"2022-11-28T12:08:32+01:00","message":"Second line", "status_code": 400}
{"level":"error","time":"2022-11-28T12:09:32+01:00","message":"Third line", "status_code": 500}
{"level":"info","time":"2022-11-28T12:10:32+01:00","message":"Forth line", "status_code": 100}
{"level":"warn","time":"2022-11-28T12:11:32+01:00","message":"Fith line", "status_code": 200}

Temos três campos:

• level: é o nível de log, queremos filtrá-lo como 'keyword' (por exemplo: 'info', 'error', 'debug', etc.).
• time: é o tempo em que a linha do log foi escrita, precisamos dizer ao Elastic-Agent para utilizá-lo como o tempo para a entrada do log, em vez do tempo de ingestão.
• message: é a mensagem em si, nós a consideraremos como um campo de texto livre.
• status_code: um campo numérico que simula um código de status HTTP.

3. Configurando a integração

Configuraremos a ' Custom Logs integration'. Além de adicionar os caminhos dos arquivos que queremos ingerir, precisamos adicionar duas configurações opcionais:

• Processors: como o nome sugere, eles podem enriquecer, modificar nossos eventos.
• Custom configurations: como o nome sugere, elas são configurações personalizadas para nossa integração.

Internamente o Elastic-Agent executará uma instância do Filebeat, portanto toda a documentação para essas configurações opcionais são a documentação do Filebeat. A integração Custom Log internamente utiliza o Log input, portanto a documentação que nos interessa:

• Processors
• JSON

A configuração personalizada que precisamos é:

Processors

Vamos precisar de dois processadores:

• 'timestamp': Ele processará o nosso timestamp e o colocará corretamente no evento final.
• 'drop_fields': Este é opcional, mas não há necessidade de manter o campo "time" se já tivermos ajustado corretamente o @timestamp no evento.

- timestamp:
    field: time
    layouts:
      - '2006-01-02T15:04:05Z07:00'
    test:
      - '2022-08-31T12:07:32+02:00'
- drop_fields:
    fields:
      - time

A única ressalva aqui é que o processador "timestamp" ainda está em beta, porém é estável o suficiente para ser utilizado. De qualquer forma, tenha isso em mente.

Custom configuration

A configuração personalizada consiste em dizer ao Log input que queremos que os dados sejam analisados como JSON, substituir quaisquer chaves que já existam no evento e se houverem erros, adiciona-los em uma chave de erro no evento final para que possamos saber o que está acontecendo.

json:
  keys_under_root: true
  add_error_key: true
  message_key: message
  overwrite_keys: true

É assim que tudo vai ficar no Kibana:

4. Testando

Salve a integração, aguarde a atualização da política propagar para seu Elastic-Agent, adicione alguns dados ao seu arquivo de log, depois vá ver os logs coletados no Kibana.

5. Mappings

Agora que temos alguns dados, vamos garantir que Elasticsearch entenda nossos dados corretamente, para isso precisamos definir os mapeamentos (mappings) dos nossos dados.

Vá até Fleet > Agent Policies, clique no nome da 'policy' e depois no nome da integração. Na tela seguinte, vá para Change defaults > Advanced options, na parte inferior há a seção Mappings*.

Clique no botão "edit" (o pequeno lápis) e adicione os seguintes mapeamentos:

Depois clique em "Next" até o passo "Review", depois clique em "Save component template".

6. Testando (novamente)

Vá para Discover, pesquise alguns dados, expanda um dos documentos e você verá os campos corretamente mapeados.

Agora você pode fazer buscas como status_code >= 400.

7. Como escolher as chaves dos meus logs?

A melhor maneira de definir suas chaves de log é usar o Elastic Common Schema (ECS). ECS é uma especificação de código aberto, desenvolvida com o apoio da comunidade de usuários Elastic. ECS define um conjunto comum de campos a serem usados no armazenamento de dados de eventos em Elasticsearch, tais como logs e métricas.

8. E o Filebeat?

Bem, é praticamente a mesma idéia, mas ao invés de configurarmos uma integração para o Elastic-Agent, configuraremos o 'input' diretamente no filebeat.yml (arquivo de configuração do Filebeat).

Ao executar um Filebeat, é melhor usar o filestream input. Os conceitos são todos iguais, mas as configurações são ligeiramente diferentes. Por uma questão de brevidade, aqui está a configuração do input no filebeat.yml:

filebeat.inputs:
- type: filestream
  id: advent-calendar-2022
  enabled: true
  paths:
    - /tmp/flog.log

  parsers:
    - ndjson:
        target: ""
        add_error_key: true

  processors:
    - timestamp:
        field: time
        layouts:
          - '2006-01-02T15:04:05Z07:00'
        test:
          - '2022-08-31T12:07:32+02:00'
    - drop_fields:
        fields: [time]

E os mapping (mapeamentos)?

Eles também podem ser editados no Kibana. Vá para Stack Management > Index Management > Index Templates, procure por filebeat. Por padrão, o Filebeat cria um data stream chamado filebeat-<versão>, portanto, no momento da escrita, temos, filebeat-8.5.2. Clique nele, depois em "manage", e "edit" no menu que aparecerá. Clique em "Next" até a seção 4, Mappings, e defina os mapeamentos como no passo 5.

O último passo é criar uma data view, vá para Stack Management > Data Views, depois clicque em "Create data view", escolha um nome e em "Index pattern", coloque "filebeat-8.5.2*". Se você não quiser usar o processador timestamp, você pode mudar aqui o campo usado como timestamp. Clique em "Save data view to Kibana".

Fica assim:

Volte para "Discover", selecione o novo data view (filebeat-8.5.2 em nosso caso) e você poderá ver todos os seus dados: