Como implementar logging centralizado com ELK Stack

1. Fundamentos do ELK Stack e Logging Centralizado

O ELK Stack é um conjunto de três ferramentas open-source que trabalham em conjunto para coletar, processar, armazenar e visualizar logs de forma centralizada. Elasticsearch armazena e indexa os dados, Logstash realiza o processamento e transformação, e Kibana oferece a interface de visualização.

O logging centralizado resolve um problema crítico em ambientes distribuídos: quando você tem dezenas ou centenas de servidores, acessar logs individualmente é inviável. Com o ELK, todos os logs convergem para um único ponto, permitindo busca em tempo real, correlação entre eventos e análises forenses detalhadas.

A arquitetura típica segue este fluxo:

Agentes (Filebeat) → Pipeline (Logstash) → Armazenamento (Elasticsearch) → Visualização (Kibana)

2. Preparação do Ambiente e Instalação dos Componentes

Pré-requisitos básicos:

- Java 11 ou 17 (OpenJDK)
- Sistema: Ubuntu 22.04 LTS ou CentOS 7+
- RAM: mínimo 4GB (recomendado 8GB para produção)
- Disco: SSD com pelo menos 50GB livres

Instalação do Elasticsearch:

# Importar chave GPG
wget -qO - https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo apt-key add -

# Adicionar repositório
echo "deb https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list

# Instalar
sudo apt-get update
sudo apt-get install elasticsearch

# Configurar cluster (single node para teste)
sudo sed -i 's/#cluster.name: my-application/cluster.name: elk-cluster/' /etc/elasticsearch/elasticsearch.yml
sudo sed -i 's/#node.name: node-1/node.name: node-1/' /etc/elasticsearch/elasticsearch.yml
sudo sed -i 's/#network.host: 192.168.0.1/network.host: 0.0.0.0/' /etc/elasticsearch/elasticsearch.yml

# Iniciar serviço
sudo systemctl enable elasticsearch
sudo systemctl start elasticsearch

Instalação do Logstash e Kibana:

sudo apt-get install logstash kibana

# Configurar Kibana
sudo sed -i 's/#server.host: "localhost"/server.host: "0.0.0.0"/' /etc/kibana/kibana.yml

sudo systemctl enable logstash
sudo systemctl enable kibana
sudo systemctl start logstash
sudo systemctl start kibana

3. Configuração do Logstash para Coleta e Processamento

Crie um pipeline básico em /etc/logstash/conf.d/meu-pipeline.conf:

input {
  beats {
    port => 5044
    ssl => false
  }
}

filter {
  # Parse de logs Nginx com grok
  if [fields][service] == "nginx" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }

    # Extrair campos específicos
    mutate {
      convert => { "[bytes]" => "integer" }
      convert => { "[response]" => "integer" }
    }

    # Adicionar timestamp do log
    date {
      match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
      target => "@timestamp"
    }
  }

  # Adicionar metadados
  mutate {
    add_field => {
      "environment" => "production"
      "application" => "web-app"
    }
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logs-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "sua-senha-aqui"
  }

  # Debug opcional
  stdout { codec => rubydebug }
}

4. Integração com Filebeat para Coleta Leve de Logs

Instale o Filebeat nos servidores de aplicação:

sudo apt-get install filebeat

Configure em /etc/filebeat/filebeat.yml:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/nginx/access.log
    - /var/log/nginx/error.log
  fields:
    service: nginx
    environment: production
  multiline:
    pattern: '^\d{2}/\w{3}/\d{4}'
    negate: true
    match: after

- type: log
  enabled: true
  paths:
    - /var/log/nodejs/app.log
  fields:
    service: nodejs
    environment: production

output.logstash:
  hosts: ["192.168.1.100:5044"]
  ssl.enabled: false

logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat.log
  keepfiles: 7

5. Configuração do Elasticsearch para Armazenamento e Indexação

Crie um template de índice para otimizar o mapeamento:

PUT _index_template/logs_template
{
  "index_patterns": ["logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "30s",
      "index.lifecycle.name": "logs_policy",
      "index.lifecycle.rollover_alias": "logs"
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "message": { "type": "text" },
        "response": { "type": "integer" },
        "bytes": { "type": "long" },
        "clientip": { "type": "ip" },
        "environment": { "type": "keyword" },
        "service": { "type": "keyword" }
      }
    }
  }
}

Configure política de ciclo de vida (ILM):

PUT _ilm/policy/logs_policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50GB",
            "max_age": "30d"
          }
        }
      },
      "warm": {
        "min_age": "30d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 }
        }
      },
      "cold": {
        "min_age": "90d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "365d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

6. Visualização e Análise com Kibana

Acesse http://seu-servidor:5601 e configure:

  1. Stack Management → Index Patterns: Crie logs-* como padrão
  2. Discover: Explore logs com filtros como service: nginx AND response: 500
  3. Dashboard: Crie visualizações combinadas

Exemplo de visualização para monitoramento de erros:

Visualização 1: Barras - Erros por serviço (últimas 24h)
  - Métrica: Count
  - Bucket: Terms (service)
  - Filtro: response >= 400

Visualização 2: Tabela - Top 10 IPs com mais erros
  - Métrica: Count
  - Bucket: Terms (clientip)

Visualização 3: Linha - Taxa de requisições por minuto
  - Métrica: Count
  - Bucket: Date Histogram (@timestamp, intervalo 1m)

7. Boas Práticas, Segurança e Manutenção

Segurança básica:

# Ativar TLS no Logstash
input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate_authorities => ["/etc/logstash/ca.crt"]
    ssl_certificate => "/etc/logstash/logstash.crt"
    ssl_key => "/etc/logstash/logstash.key"
  }
}

# Criar usuários no Elasticsearch
POST _security/user/logstash_writer
{
  "password": "senha-segura",
  "roles": ["logstash_writer"]
}

Logging estruturado (formato JSON):

# Exemplo de log JSON para Node.js
{"level":"error","message":"Falha na conexão com banco","timestamp":"2025-01-15T10:30:00Z","service":"nodejs","requestId":"abc123","error":"ECONNREFUSED"}

Monitoramento do cluster:

GET _cluster/health
GET _cat/nodes?v
GET _cat/indices?v&s=index:desc

8. Troubleshooting e Estudos de Caso

Problema comum: Logstash não processa logs

# Verificar se o pipeline está ativo
sudo systemctl status logstash

# Testar pipeline manualmente
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/test.conf --config.test_and_exit

# Debug com stdout
output {
  stdout { codec => rubydebug }
}

Caso prático: Centralizando logs de Nginx + Node.js + PostgreSQL

# Arquivo de configuração final do Logstash
input {
  beats {
    port => 5044
  }
}

filter {
  if [fields][service] == "nginx" {
    grok { match => { "message" => "%{COMBINEDAPACHELOG}" } }
  }
  else if [fields][service] == "nodejs" {
    json { source => "message" }
  }
  else if [fields][service] == "postgresql" {
    grok {
      match => { 
        "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level}: %{GREEDYDATA:sql_query}"
      }
    }
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logs-%{[fields][service]}-%{+YYYY.MM.dd}"
  }
}

O logging centralizado com ELK Stack transforma dados brutos em insights acionáveis. Comece com um ambiente simples, valide cada componente e evolua gradualmente para cenários mais complexos.

Referências