A importância da gestão eficiente de logs no Elasticsearch
No ecossistema moderno de infraestrutura e observabilidade, o Elasticsearch consolidou-se como a espinha dorsal para armazenamento e análise de dados de log. Ferramentas como o Elastic Stack (ELK) permitem que equipes de SRE, DevOps e desenvolvedores monitorem a saúde de aplicações em tempo real. No entanto, à medida que o volume de dados cresce exponencialmente, a estratégia padrão de indexação por data — criar um novo índice diariamente ou mensalmente — torna-se insuficiente.
O principal problema reside na fragmentação. Centenas de índices pequenos geram uma sobrecarga significativa no cluster, consumindo recursos valiosos de CPU e memória para gerenciamento de metadados, além de impactar a performance das buscas distribuídas. Para mitigar isso, o Elasticsearch oferece o recurso nativo de Rollover. Diferente da rotação baseada em tempo, o rollover permite que um índice cresça até atingir critérios específicos, como tamanho máximo ou número de documentos, antes de ser fechado e substituído por um novo.
Neste tutorial técnico, demonstraremos como implementar uma automação robusta para o rollover de logs utilizando scripts em Python. Abordaremos a configuração do Index Lifecycle Management (ILM), a criação de templates de índice e a escrita de um script automatizado que monitora o estado dos índices e executa a transição necessária, garantindo uma infraestrutura escalável e performática.
Pré-requisitos e arquitetura da solução
Antes de iniciar a implementação, é fundamental entender os componentes envolvidos na automação do rollover. A solução não depende apenas do script em Python; ela requer uma configuração prévia no cluster Elasticsearch para definir as regras de negócio.
- Elasticsearch Cluster: Versão 7.x ou superior recomendada. Certifique-se de ter acesso administrativo com privilégios para gerenciar índices e políticas ILM.
- Python 3.8+: Linguagem escolhida por sua simplicidade e vasta biblioteca de suporte para APIs REST.
- Biblioteca
elasticsearch-py: O cliente oficial Python para o Elasticsearch, que facilita a comunicação com a API REST. - Elasticsearch Index Template: Estrutura que define como os novos índices serão criados após o rollover.
- ILM Policy (Política de Ciclo de Vida): Define o que acontece com o índice após ele ser fechado (ex: deletar após 30 dias, mover para cold storage).
A arquitetura proposta segue um fluxo contínuo: um índice primário (logs-000001) recebe todas as escritas. Quando atinge o limite definido, o sistema executa o rollover, fecha o índice antigo e abre um novo (logs-000002). O script em Python atuará como o gatilho agendado (via cron ou sistema de orquestração) para verificar essa condição.
Passo 1: Configurar a Política ILM e o Template de Índice
O primeiro passo é garantir que o Elasticsearch saiba como lidar com os índices antigos e novos. Vamos criar uma política ILM chamada logs_policy que move o índice para o estado Closed após 1 dia (para economizar recursos durante a retenção curta) e depois o deleta após 7 dias. Em seguida, criamos um template de índice que aplica essa política automaticamente.
Execute os seguintes comandos via curl ou no Kibana Dev Tools para configurar a infraestrutura base:
# 1. Criar a política ILM
PUT _ilm/policy/logs_policy
{
"policy": {
"phases": {
"hot": {
"actions": {
"rollover": {
"max_age": "1d",
"max_docs": 10000,
"max_size": "5gb"
}
}
},
"delete": {
"min_age": "7d",
"actions": {
"delete": {}
}
}
}
}
}Em seguida, definimos o template de índice que será aplicado a todos os índices que começarem com logs-. Note a propriedade is_write_index: true, que indica ao Elasticsearch qual índice deve receber as novas escritas.
# 2. Criar o Index Template
PUT _index_template/logs_template
{
"index_patterns": ["logs-*"],
"priority": 100,
"template": {
"settings": {
"number_of_shards": 1,
"number_of_replicas": 1,
"index.lifecycle.name": "logs_policy",
"index.lifecycle.rollover_alias": "logs_write_alias"
},
"mappings": {
"properties": {
"@timestamp": {
"type": "date"
},
"message": {
"type": "text"
},
"level": {
"type": "keyword"
}
}
}
}
}A chave aqui é o index.lifecycle.rollover_alias. Ao invés de gerenciar nomes de índices específicos, apontamos para um alias. Isso permite que o script e as aplicações escrevam sempre no mesmo endereço lógico, enquanto o Elasticsearch gerencia a troca interna do índice físico.
Passo 2: Inicializar o Índice Primário
Para que o rollover funcione, é necessário criar o primeiro índice manualmente ou via script. Este índice servirá como o ponto de partida. Vamos nomeá-lo logs-000001.
# Criar o índice inicial
PUT logs-000001
{
"aliases": {
"logs_write_alias": {
"is_write_index": true
}
}
}Agora, qualquer dado enviado para o alias logs_write_alias será armazenado no índice logs-000001. Verifique se o índice está ativo:
GET logs_write_alias/_search
{
"size": 1
}Passo 3: Desenvolver o Script de Automação em Python
Agora que a infraestrutura está pronta, precisamos do agente de automação. O script verificará se o índice atual atingiu os limites definidos na política ILM e, caso positivo, solicitará o rollover.
Primeiro, instale a dependência necessária:
pip install elasticsearchAbaixo, apresentamos o código completo do script elastic_rollover.py. Este script conecta ao cluster, verifica o estado do índice de escrita e executa a operação se necessário.
import json
import sys
from elasticsearch import Elasticsearch
def check_and_rollover(es_client, alias_name="logs_write_alias"):
"""
Verifica o status do índice apontado pelo alias e realiza rollover se necessário.
"""
try:
# 1. Obter informações sobre os índices vinculados ao alias
indices_info = es_client.indices.get_alias(name=alias_name)
if not indices_info:
print(f"Erro: Alias '{alias_name}' não encontrado.")
return False
# Identificar o índice atual que é o write_index
current_index = None
for index_name, info in indices_info.items():
aliases = info.get("aliases", {})
if alias_name in aliases and aliases[alias_name].get("is_write_index", False):
current_index = index_name
break
if not current_index:
print(f"Erro: Nenhum índice write found para o alias '{alias_name}'.")
return False
# 2. Verificar os limites do ILM
# Precisamos verificar se o rollover foi solicitado ou se as condições foram atingidas.
# O Elasticsearch não retorna explicitamente "pronto para rollover" via API simples,
# mas podemos checar o status do ILM ou tentar executar o comando e lidar com o erro.
# Abordagem recomendada: Tentar verificar o status do índice no contexto do ILM
# Ou simplesmente tentar o rollover. Se já estiver pronto, o ES gerencia.
# No entanto, para logs limpos, vamos checar o tamanho/docs se possível via stats.
stats = es_client.indices.stats(index=current_index)
store_stats = stats["_indices"][current_index]["store"]
size_in_bytes = store_stats["size_in_bytes"]
doc_count = stats["_indices"][current_index]["docs"]["count"]
# Limites definidos na política (exemplo: 5GB ou 10k docs)
MAX_SIZE_BYTES = 5 * 1024 * 1024 * 1024 # 5 GB
MAX_DOCS = 10000
print(f"Índice atual: {current_index}")
print(f"Tamanho: {size_in_bytes / (1024**3):.2f} GB")
print(f"Documentos: {doc_count}")
# Verificar condições de rollover
should_rollover = False
if size_in_bytes >= MAX_SIZE_BYTES:
print("Condição de tamanho atingida. Iniciando rollover...")
should_rollover = True
if doc_count >= MAX_DOCS:
print("Condição de documentos atingida. Iniciando rollover...")
should_rollover = True
# Nota: A condição de tempo (max_age) é gerenciada internamente pelo ILM,
# mas o script pode forçar o rollover se quiser basear-se apenas em tamanho/docs.
# Para uma automação completa via ILM, muitas vezes não é necessário rodar este script
# se o agente do Elasticsearch estiver ativo, mas para cenários externos ou
# triggers customizadas, este check é vital.
if should_rollover:
# 3. Executar o Rollover
print(f"Executando rollover no índice {current_index}...")
response = es_client.indices.rollover(alias=alias_name)
if response.get("acknowledged"):
new_index = list(response["indices"].keys())[0] if current_index in response["indices"] else "Desconhecido"
print(f"Scesso! Novo índice criado ou rollover realizado.")
return True
else:
print(f"Falha no rollover: {response}")
return False
else:
print("Condições de rollover não atingidas. Nada a fazer.")
return True
except Exception as e:
print(f"Erro ao processar rollover: {str(e)}")
return False
if __name__ == "__main__":
# Configuração de conexão
ES_HOST = "localhost"
ES_PORT = 9200
es = Elasticsearch([{"host": ES_HOST, "port": ES_PORT}])
# Verificar conectividade básica
if not es.ping():
print("Não foi possível conectar ao Elasticsearch.")
sys.exit(1)
# Executar a lógica
check_and_rollover(es)Passo 4: Implementação em Produção e Agendamento
Com o script funcional, o próximo passo é integrá-lo ao ambiente de produção. Recomenda-se não rodar este script manualmente, mas sim através de um agendador de tarefas.
Opção A: Cron Job (Linux)
Adicione uma entrada no crontab do usuário responsável para verificar a cada 15 minutos:
/15 * * * * /usr/bin/python3 /opt/scripts/elastic_rollover.py >> /var/log/elastic_rollover.log 2>&1Opção B: Sistema de Orquestração (Kubernetes/Celery)
Se sua infraestrutura estiver em Kubernetes, utilize um CronJob. Isso permite melhor gerenciamento de logs e falhas. Se você utiliza Python para outras automações, considere usar o Airflow ou Celery para criar uma tarefa agendada que invoca a função check_and_rollover.
Considerações de Segurança:
- Nunca armazene credenciais em texto puro no script. Utilize variáveis de ambiente ou ferramentas como AWS Secrets Manager, HashiCorp Vault ou configurações do próprio Elasticsearch com autenticação básica/NTLM.
- Configure o script para enviar alertas (via Slack, PagerDuty ou Email) caso a conexão falhe ou o rollover seja interrompido por erro de permissão.
Boas Práticas e Otimizações Avançadas
A automação do rollover é poderosa, mas exige monitoramento contínuo. Aqui estão dicas cruciais para manter a saúde do seu cluster:
- Mapeamentos Dinâmicos: Certifique-se de que o template de índice proíba a adição dinâmica de campos não mapeados em índices antigos. Isso evita que novos tipos de logs quebrem buscas existentes. Use
"dynamic": "strict"no mapeamento. - Sharding Estratégico: Ao definir o número de shards, lembre-se de que um rollover cria um novo índice com seus próprios shards. Se você criar milhares de índices pequenos por dia, o cluster entrará em colapso. Use
max_ageoumax_docspara limitar a frequência de criação. - Monitoramento do ILM: Utilize a API
GET _ilm/explainpara verificar se os índices estão seguindo as políticas corretamente. Erros no ILM são comuns quando há problemas de disco ou permissão. - Alias Naming Convention: Mantenha uma convenção rigorosa de nomes para aliases. Use prefixos claros como
-write,-reade-searchpara evitar ambiguidades em clusters grandes.
Conclusão
A automação do rollover de logs no Elasticsearch não é apenas uma questão de conveniência, mas uma necessidade técnica para escalabilidade. Ao substituir a rotação baseada puramente em tempo por critérios baseados em dados (tamanho e volume), você otimiza o uso de recursos e simplifica a gestão do ciclo de vida dos seus dados.
A combinação de Index Templates, ILM Policies e scripts em Python oferece um controle granular e programável sobre sua infraestrutura de logging. Este tutorial forneceu a base sólida para que sysadmins e desenvolvedores possam implementar essa automação, garantindo que o Elasticsearch permaneça rápido, estável e pronto para lidar com o crescimento exponencial dos dados da organização.
Lembre-se: teste sempre em um ambiente de staging antes de aplicar políticas de deleção ou fechamento automático em produção. A visibilidade é a chave para uma infraestrutura resiliente.