Integração da Stack ELK: Sincronizando Logstash, Elasticsearch e Kibana

Quando Logstash, Elasticsearch e Kibana estão dessincronizados, logs desaparecem, dashboards ficam vazios ou campos aparecem com o tipo errado. A integração da Stack ELK é menos sobre iniciar três serviços e mais sobre garantir que os nomes de índices, mapeamentos, timestamps, credenciais e visualizações de dados estejam todos alinhados.

Este guia percorre um pipeline de logging prático: Logstash recebe eventos, analisa-os, envia-os para o Elasticsearch, e o Kibana lê os índices ou fluxos de dados resultantes. Os exemplos usam sintaxe clássica de pipeline do Logstash e APIs do Elasticsearch que você pode executar no Kibana Dev Tools.

Entendendo o Fluxo de Dados

Trace um evento através da stack:

1. Logstash recebe dados de Beats, TCP, syslog, arquivos, filas ou outra entrada.
2. Filtros do Logstash analisam, enriquecem, renomeiam e normalizam campos.
3. Elasticsearch indexa o evento usando templates, mapeamentos e políticas de ciclo de vida.
4. Kibana consulta o Elasticsearch através de uma visualização de dados e exibe o evento no Discover, dashboards, Lens ou alertas.

A maioria dos bugs de integração ocorre nos limites. Logstash não consegue conectar, Elasticsearch rejeita o documento, ou Kibana está olhando para a visualização de dados ou intervalo de tempo errados.

Configuração do Logstash para Fluxo de Dados Limpo

Pipelines do Logstash têm três blocos principais: input, filter e output. Mantenha cada bloco simples e testável.

Plugins de Entrada

Plugins de entrada comuns incluem:

• beats: Recebe eventos do Filebeat, Metricbeat e outros Beats.
• tcp / udp: Recebe eventos através de sockets de rede.
• file: Lê arquivos locais. Isso é útil para pequenas implantações e testes, mas agentes geralmente são melhores para hosts de produção distribuídos.
• syslog: Recebe mensagens syslog.

Exemplo de entrada Beats com TLS:

input {
  beats {
    port => 5044
    ssl_enabled => true
    ssl_certificate => "/etc/pki/tls/certs/logstash.crt"
    ssl_key => "/etc/pki/tls/private/logstash.key"
  }
}

Certifique-se de que a porta está aberta, o certificado corresponde a como os clientes se conectam e os nomes das opções correspondem à versão do plugin instalado. Versões recentes do plugin de entrada Beats usam ssl_enabled.

Plugins de Filtro

Filtros transformam eventos brutos em campos úteis. A ordem importa porque o Logstash executa filtros sequencialmente.

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }

  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }

  mutate {
    remove_field => [ "message" ]
  }
}

Use grok para texto não estruturado, date para definir @timestamp, mutate para limpeza de campos e geoip quando precisar de enriquecimento de localização baseado em IP. Teste padrões grok contra linhas de log reais antes de colocá-los em produção. Um pequeno erro de análise pode enviar milhares de eventos para o Elasticsearch com campos ausentes.

Plugin de Saída

Para a stack ELK, a saída Elasticsearch é o destino usual.

output {
  elasticsearch {
    hosts => ["https://elasticsearch-node1:9200", "https://elasticsearch-node2:9200"]
    index => "my-logs-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${LOGSTASH_ES_PASSWORD}"
    ssl_enabled => true
    cacert => "/etc/logstash/certs/http_ca.crt"
  }
}

O valor index é o contrato com os templates do Elasticsearch e as visualizações de dados do Kibana. Se o Logstash escrever my-logs-2026.05.23, seu template e visualização de dados devem corresponder a my-logs-*.

Para ambientes maiores, considere fluxos de dados e Gerenciamento de Ciclo de Vida de Índices em vez de índices diários gerenciados manualmente. Se você usar fluxos de dados, siga as orientações atuais do Elastic para saída do Logstash para configurações data_stream em vez de misturar opções de fluxo de dados e índices clássicos.

Templates e Mapeamentos do Elasticsearch

O Elasticsearch precisa de mapeamentos consistentes antes que os documentos cheguem. Caso contrário, o primeiro documento pode definir um tipo de campo que quebra eventos posteriores. Um código de status que chega primeiro como "200" pode se tornar texto ou keyword em vez de um número.

Exemplo de template de índice componível:

PUT _index_template/my_log_template
{
  "index_patterns": ["my-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1
    },
    "mappings": {
      "properties": {
        "@timestamp": {"type": "date"},
        "message": {"type": "text"},
        "host.name": {"type": "keyword"},
        "log.level": {"type": "keyword"},
        "http.response.status_code": {"type": "integer"}
      }
    }
  }
}

Use keyword para correspondência exata e agregações, text para pesquisa de texto completo, tipos numéricos para métricas e códigos de status, e date para campos de tempo. Mantenha contagens de shards modestas, a menos que você tenha uma razão medida para adicionar mais. Muitos shards pequenos podem prejudicar o desempenho do cluster.

Visualizações de Dados do Kibana

A interface atual do Kibana usa visualizações de dados. Versões mais antigas chamavam-nas de padrões de índice. Crie uma visualização de dados que corresponda aos nomes de índices ou fluxos de dados que o Elasticsearch realmente tem.

Configuração típica:

1. Vá para Stack Management -> Kibana -> Data Views.
2. Crie uma visualização de dados como my-logs-*.
3. Escolha @timestamp como o campo de tempo.
4. Abra o Discover e amplie o seletor de tempo durante os testes.

Se o Discover estiver vazio, não assuma que o Logstash falhou. Verifique o intervalo de tempo, o padrão da visualização de dados e se @timestamp foi analisado corretamente.

Solucionando Problemas Comuns de Integração

Dados Não Aparecem no Kibana

Verifique cada etapa:

GET _cat/indices/my-logs-*?v
GET my-logs-*/_search?size=1&sort=@timestamp:desc

Depois verifique:

• Logs do Logstash para erros de conexão, autenticação, TLS ou mapeamento.
• Logs do Elasticsearch para documentos rejeitados e falhas de segurança.
• O padrão da visualização de dados do Kibana e o intervalo de tempo selecionado.
• Se o timestamp do evento está no futuro, no passado ou ausente.

Documentos São Rejeitados pelo Elasticsearch

Conflitos de mapeamento são comuns. Por exemplo, um evento envia http.response.status_code como 200, enquanto outro envia "OK". O Elasticsearch não pode armazenar ambos em um campo integer.

Corrija o filtro do Logstash para que o campo seja tipado consistentemente, ou encaminhe eventos ruins para um índice separado para revisão. Não fique deletando e recriando índices sem corrigir o pipeline que cria os documentos ruins.

Logstash Usa Muita CPU

Padrões grok caros, alto volume de eventos e eventos multiline grandes podem aumentar rapidamente a CPU do Logstash. Comece medindo qual pipeline está ocupado, depois simplifique padrões, ancore regexes e mova a análise simples para pipelines de ingestão do Beats ou Elasticsearch quando for mais fácil de operar.

Consultas do Kibana Estão Lentas

Dashboards lentos geralmente vêm de intervalos de tempo amplos, agregações de alta cardinalidade, muitos shards ou campos mapeados como text quando o Kibana precisa de keyword. Use padrões de dashboard mais restritos, rollover do ILM e mapeamentos de campo que correspondam às suas visualizações.

Conclusão

Trate a integração da Stack ELK como um contrato entre três camadas. O Logstash deve emitir campos previsíveis, o Elasticsearch deve mapeá-los e armazená-los corretamente, e o Kibana deve consultar a visualização de dados certa no intervalo de tempo certo. Quando algo quebrar, siga um evento de amostra desde a entrada até o índice e o dashboard.