Log Aggregation

Em um sistema distribuído com dezenas de serviços, cada serviço gera logs em seu próprio nó ou container. Para investigar um incidente, você precisaria acessar manualmente cada container — inviável em escala e impossível após o container ser removido. A agregação de logs resolve isso centralizando todos os logs em um único sistema pesquisável com retenção configurável. A escolha do sistema de aggregation não é técnica apenas: ela define o custo de storage do time, a velocidade de debugging, e a complexidade operacional que você vai carregar.

Requisitos de um Sistema de Log Aggregation

• Coleta confiável: garantia de que logs não são perdidos mesmo sob alto volume ou indisponibilidade temporária do backend. Buffer em disco no agente é o mecanismo padrão.
• Baixa latência de ingestão: logs devem aparecer no sistema central em segundos (não minutos) para uso em debugging ativo durante incidentes.
• Busca eficiente: consultas por conteúdo, campo, time range sem full-scan em terabytes de dados.
• Custo controlado: logs são o sinal de observabilidade mais volumoso — custo de storage e ingestão pode facilmente superar métricas e traces combinados.
• Retenção configurável: hot storage para investigação ativa (7-30 dias), cold storage para compliance e auditoria (1-7 anos).
• Controle de acesso: logs podem conter dados sensíveis — RBAC por serviço e ambiente, com audit trail de quem buscou o quê.

ELK Stack (Elasticsearch + Logstash + Kibana)

O stack ELK (hoje chamado Elastic Stack) é o sistema de log aggregation mais maduro e amplamente adotado. Cada componente tem uma função específica na pipeline de ingestão, processamento e visualização.

Elasticsearch

Motor de busca e analytics baseado em Lucene. Armazena logs como documentos JSON e cria índices invertidos para busca eficiente por qualquer campo. Características críticas em produção:

• Sharding e replicação: índices são divididos em shards distribuídos por nós. Réplicas garantem disponibilidade e leitura paralela. Over-sharding é um erro comum — mais shards não significa mais performance se os shards são pequenos.
• Index Lifecycle Management (ILM): automatiza o ciclo de vida dos índices — hot (SSD, alta frequência de busca), warm (HDD, busca ocasional), cold (object storage, acesso raro), delete. Reduz custo significativamente sem intervenção manual.
• Mapping: o schema dos documentos. Mappings incorretos (ex: um campo às vezes string, às vezes int) causam "mapping explosions" — criação de dezenas de milhares de campos dinâmicos que degradam performance e podem desestabilizar o cluster.

Logstash vs Beats vs Fluent Bit

Logstash: processamento pesado (parse, transform, enrich) mas alto uso de memória e CPU — adequado para pipelines complexas de transformação. Beats (Filebeat, Metricbeat): agentes leves para coleta específica. Fluent Bit: ultra-leve (~450KB de binário, ~5MB de RAM), baixíssima latência, preferido em Kubernetes — um DaemonSet por nó que coleta logs de todos os containers do nó e encaminha para qualquer backend.

Kibana e KQL

UI de visualização e busca. KQL (Kibana Query Language) para busca ad-hoc — mais simples que Lucene syntax para a maioria dos casos. Dashboards, alertas, e desde a versão 8, integração com ML para detecção de anomalias em logs.

// Elasticsearch — ILM policy para logs de produção
PUT _ilm/policy/logs-production
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50gb",  // novo índice ao atingir 50GB
            "max_age": "1d"      // ou após 1 dia
          },
          "set_priority": { "priority": 100 }
        }
      },
      "warm": {
        "min_age": "3d",         // após 3 dias no hot
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 },
          "set_priority": { "priority": 50 }
        }
      },
      "cold": {
        "min_age": "14d",
        "actions": { "freeze": {} }  // libera heap, acesso mais lento
      },
      "delete": {
        "min_age": "30d",
        "actions": { "delete": {} }
      }
    }
  }
}

// KQL — exemplos de busca no Kibana
// Todos os erros de um serviço:
service.name: "order-service" AND level: "ERROR"

// Busca por trace ID (correlação com traces):
trace_id: "4bf92f3577b34da6a3ce929d0e0e4736"

// Full-text + filtros:
message: "payment failed" AND status_code: 500 AND NOT user_id: *

Grafana Loki

Loki foi projetado com uma filosofia radicalmente diferente do Elasticsearch: não indexe o conteúdo dos logs, apenas os labels. Isso reduz dramaticamente o custo de storage e a complexidade operacional — a troca é que busca full-text é mais lenta (Loki precisa escanear chunks não indexados dentro de cada stream selecionado pelos labels).

Modelo de dados: Streams e Labels

No Loki, logs são organizados em streams — conjuntos de logs com exatamente os mesmos labels. Labels são as dimensões indexadas: service, environment, pod, namespace. O conteúdo do log (a mensagem em si) não é indexado — é armazenado como chunks comprimidos em object storage (S3, GCS).

LogQL

LogQL é a query language do Loki, com sintaxe inspirada em PromQL. Dois tipos de queries: log queries (retornam linhas de log filtradas) e metric queries (derivam métricas dos logs via funções como rate() e quantile_over_time()).

// LogQL — exemplos práticos

// Log query: selecionar stream + filtrar + parsear + reformatar
{service="order-service", environment="production"}
  |= "ERROR"                    // filtro de substring
  != "health"                   // excluir linhas com "health"
  | json                        // parsear linha como JSON
  | level="error"               // filtrar campo JSON parseado
  | line_format "{{.message}} traceId={{.trace_id}}"

// Filtros com campos parseados
{service="order-service"}
  | json
  | duration > 1s               // filtrar por duração
  | status_code >= 500

// Metric query: error rate derivada de logs (sem Prometheus)
sum(rate(
  {service="order-service"} |= "ERROR" [5m]
)) by (service)

// Latência P99 se logs têm campo duration
quantile_over_time(0.99,
  {service="order-service"}
  | json
  | unwrap duration [5m]
) by (service)

// Parse via regexp para logs não-JSON (ex: nginx)
{service="nginx"}
  | regexp `(?P<method>\w+) (?P<path>/[^\s]+) HTTP/[^\s]+ (?P<status>\d+)`
  | status >= "500"

// Buscar trace_id para correlação traces→logs
{service="order-service"}
  |= "4bf92f3577b34da6a3ce929d0e0e4736"

# Arquitetura Loki em produção (loki-values.yaml / Helm)
loki:
  storage:
    type: s3                    # object storage — custo muito baixo
    s3:
      bucketnames: my-loki-bucket
      region: us-east-1

  schemaConfig:
    configs:
      - from: 2024-01-01
        store: tsdb             # TSDB store — mais eficiente
        object_store: s3
        schema: v13
        index:
          prefix: loki_index_
          period: 24h

  limits_config:
    ingestion_rate_mb: 16
    ingestion_burst_size_mb: 32
    max_streams_per_user: 10000 # limite de cardinalidade por tenant
    retention_period: 744h      # 31 dias

# Modo microservices para alta escala:
# - Distributor: recebe logs, valida, hash por stream
# - Ingester: acumula chunks em memória, flushia para S3
# - Querier: lê de S3 para queries históricas
# - Query Frontend: cache, paralelização de queries grandes

CloudWatch Logs (AWS)

CloudWatch Logs é o sistema de log aggregation gerenciado da AWS. Integração nativa com todos os serviços AWS (Lambda, ECS, EKS via Fluent Bit, API Gateway, ALB, RDS). Zero operação — você paga por ingestão e armazenamento. Ideal para workloads AWS-native onde a integração automática justifica o custo.

• Log Groups: agrupamentos de logs por serviço/aplicação. Cada Log Group tem retenção configurável independente (de 1 dia a 10 anos).
• Log Streams: sequência de log events dentro de um Log Group — geralmente um por instância ou container.
• Metric Filters: extraem métricas de logs sem instrumentar o código — ex: contar linhas com "ERROR" e criar uma CloudWatch Metric automaticamente.
• Insights: linguagem de query SQL-like para análise interativa com suporte a agregações, regex parse e stats.
• S3 Export + Athena: para queries históricas de longa duração, exporte para S3 e query via Athena (serverless SQL).

// CloudWatch Logs Insights — queries práticas

// Distribuição de erros por endpoint (últimas 24h)
fields @timestamp, @message
| filter @message like /ERROR/
| parse @message "path=* status=*" as path, status
| stats count(*) as error_count by path
| sort error_count desc
| limit 20

// Latência P95 por endpoint
fields @timestamp, @message
| parse @message '"path":"*","duration":*' as path, duration_ms
| filter ispresent(duration_ms)
| stats pct(duration_ms, 95) as p95_ms by path
| sort p95_ms desc

// Buscar trace_id específico
fields @timestamp, @message
| filter @message like "4bf92f3577b34da6a3ce929d0e0e4736"
| sort @timestamp asc

// Análise de cold starts Lambda
filter @type = "REPORT"
| stats
    avg(initDuration) as avg_cold_start,
    count(initDuration) as cold_start_count,
    count(*) as total_invocations
  by bin(1h)

Comparação: ELK vs Loki vs CloudWatch

Configuração de Coleta por Linguagem

// Packages:
// Serilog.Sinks.Elasticsearch
// Serilog.Sinks.Grafana.Loki
// Serilog.Enrichers.Environment

// appsettings.json — configuração declarativa
{
  "Serilog": {
    "MinimumLevel": {
      "Default": "Information",
      "Override": {
        "Microsoft.AspNetCore": "Warning",
        "Microsoft.EntityFrameworkCore.Database.Command": "Warning"
      }
    },
    "Enrich": ["FromLogContext", "WithMachineName", "WithProcessId"],
    "WriteTo": [
      {
        "Name": "Elasticsearch",
        "Args": {
          "nodeUris": "http://elasticsearch:9200",
          "indexFormat": "app-logs-{0:yyyy.MM.dd}",
          "autoRegisterTemplate": true,
          "autoRegisterTemplateVersion": "ESv8",
          "period": "00:00:05",
          "bufferBaseFilename": "/var/log/app/buffer"
        }
      },
      {
        "Name": "GrafanaLoki",
        "Args": {
          "uri": "http://loki:3100",
          "labels": [
            { "key": "service", "value": "order-service" },
            { "key": "environment", "value": "production" }
          ],
          "batchPostingLimit": 1000,
          "period": "00:00:02"
        }
      }
    ]
  }
}

// Program.cs — com trace_id automático via OTel bridge
Log.Logger = new LoggerConfiguration()
    .ReadFrom.Configuration(configuration)
    .Enrich.WithProperty("service.name", "order-service")
    .Enrich.WithProperty("service.version",
        Assembly.GetExecutingAssembly().GetName().Version?.ToString())
    .Enrich.WithOpenTelemetry()  // injeta trace_id e span_id automaticamente
    .CreateLogger();

import structlog
import logging
from opentelemetry import trace

def add_trace_context(logger, method, event_dict):
    span = trace.get_current_span()
    if span and span.is_recording():
        ctx = span.get_span_context()
        event_dict["trace_id"] = format(ctx.trace_id, "032x")
        event_dict["span_id"] = format(ctx.span_id, "016x")
    return event_dict

def drop_health_check_logs(logger, method, event_dict):
    # não logar health checks — reduz volume sem valor observacional
    if event_dict.get("path") in ("/health", "/metrics", "/readiness"):
        raise structlog.DropEvent()
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        add_trace_context,
        drop_health_check_logs,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),  # stdout JSON
    ],
    wrapper_class=structlog.BoundLogger,
    logger_factory=structlog.PrintLoggerFactory(),
)

# Em Kubernetes: logs vão para stdout → Fluent Bit DaemonSet
# captura automaticamente, adiciona metadados k8s, envia para Loki.
# O structlog só precisa escrever JSON em stdout — coleta é responsabilidade
# da infraestrutura, não da aplicação.

// logger/setup.go — slog com handler JSON + trace injection
package logger

import (
    "context"
    "log/slog"
    "os"
    "go.opentelemetry.io/otel/trace"
)

type traceHandler struct{ slog.Handler }

func (h *traceHandler) Handle(ctx context.Context, r slog.Record) error {
    span := trace.SpanFromContext(ctx)
    if span.IsRecording() {
        sCtx := span.SpanContext()
        r.AddAttrs(
            slog.String("trace_id", sCtx.TraceID().String()),
            slog.String("span_id", sCtx.SpanID().String()),
        )
    }
    return h.Handler.Handle(ctx, r)
}

func New(serviceName, version string) *slog.Logger {
    base := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelInfo,
        ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr {
            if a.Key == slog.MessageKey { a.Key = "message" }
            return a
        },
    })
    return slog.New(&traceHandler{base}).With(
        slog.String("service.name", serviceName),
        slog.String("service.version", version),
        slog.String("environment", os.Getenv("ENV")),
    )
}

// fluent-bit DaemonSet config (values.yaml / Helm)
// [INPUT]
//   Name              tail
//   Path              /var/log/containers/*.log
//   multiline.parser  docker, cri
//   Tag               kube.*
//
// [FILTER]
//   Name       kubernetes
//   Match      kube.*
//   Merge_Log  On   # merge JSON do container com metadados k8s
//
// [FILTER]
//   Name     grep
//   Match    kube.*
//   Exclude  $kubernetes['container_name'] fluent-bit
//
// [OUTPUT]
//   Name        loki
//   Match       kube.*
//   Host        loki.monitoring.svc.cluster.local
//   Port        3100
//   Labels      job=fluentbit, cluster=prod
//   Label_Keys  $kubernetes['namespace_name'],$kubernetes['pod_name']
//   Remove_Keys kubernetes,stream

Como praticar

1. Configure um pipeline completo de log aggregation com Fluent Bit + Loki em Docker Compose: Fluent Bit coletando logs de containers, adicionando labels de low cardinality (service, environment), e enviando para Loki. Visualize no Grafana com pelo menos duas queries LogQL — uma log query com filtro e parse JSON, e uma metric query derivando error rate dos logs.
   Critério: logs dos containers aparecem no Loki em menos de 10 segundos; a log query retorna logs filtrados por nível e serviço corretamente; a metric query de error rate produz um gráfico de série temporal válido no Grafana; labels usados são todos de baixa cardinalidade (nenhum user_id ou request_id como label).
2. Demonstre o impacto de alta cardinalidade no Loki: configure dois pipelines — um com labels de baixa cardinalidade (service, environment) e outro adicionando user_id como label. Compare o número de streams criados, o uso de memória do Loki, e o tempo de query entre as duas configurações. Documente por que a segunda abordagem é problemática.
   Critério: o pipeline com user_id como label cria N streams (onde N = número de usuários distintos); o pipeline correto tem ≤ 5 streams para o mesmo conjunto de logs; o tempo de query do pipeline correto é menor; o documento explica como buscar por user_id no conteúdo via LogQL (| json | user_id="123") sem usar label.
3. Configure um Elasticsearch com ILM policy: crie a policy com fases hot (rollover em 10GB ou 1 dia), warm (shrink para 1 shard após 3 dias), cold (após 7 dias), delete (após 14 dias). Crie um índice de teste, ingira logs fictícios, e valide que o ILM move o índice pelas fases corretas. Escreva 3 queries KQL que você usaria durante um incidente.
   Critério: a ILM policy está configurada com todas as 4 fases; o índice de teste move para warm após o rollover ser forçado manualmente (POST index/_rollover); as 3 queries KQL incluem: busca por level+service, busca por trace_id específico, e busca por mensagem com filtro de time range; o explain da ILM mostra a fase atual e próxima ação.
4. Implemente filtragem de logs antes da ingestão no Fluent Bit para reduzir volume: configure filtros para descartar logs de health check (/health, /readiness, /metrics), logs de nível DEBUG, e logs de um bot específico por User-Agent. Meça o volume de logs antes e depois dos filtros com a métrica fluentbit_output_proc_bytes_total.
   Critério: logs de health check não aparecem no Loki/ES; logs DEBUG não aparecem em produção; o volume reduzido é mensurável (pelo menos 20% de redução para uma app típica); a configuração dos filtros está documentada com justificativa para cada filtro.
5. Configure correlação logs→traces em um stack Grafana+Loki+Tempo: injete trace_id nos logs da aplicação (via OTel Logs Bridge ou manualmente), configure "derived fields" no datasource Loki que reconhece o padrão de trace_id e gera link para o Tempo, e valide a navegação de um log específico para o trace correspondente.
   Critério: logs contêm o campo trace_id no formato hex de 32 caracteres; o Grafana detecta o campo e exibe botão "View Trace" no Loki Explore; clicar no botão abre o trace correto no Tempo; a navegação funciona mesmo para trace_ids de spans aninhados (não só o span raiz).