Vector: pipeline de logs self-hosted

# docker-compose.yml — Vector como agregador de logs:
services:
  vector:
    image: timberio/vector:latest-alpine
    restart: always
    volumes:
      - ./vector.toml:/etc/vector/vector.toml:ro
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - /var/run/docker.sock:/var/run/docker.sock
      - /var/log:/var/log:ro
    environment:
      LOKI_URL: http://loki:3100
    ports:
      - "127.0.0.1:8686:8686"  # API de status do Vector

# Verificar status:
curl http://localhost:8686/health
curl http://localhost:8686/components   # ver fontes/destinos ativos

# vector.toml — pipeline Docker → Loki

# ── Fontes (sources) ─────────────────────────────────────
[sources.docker_logs]
type = "docker_logs"
docker_host = "unix:///var/run/docker.sock"
# Coletar de todos os containers:
# include_containers = []
# Ou apenas alguns:
# include_containers = ["minha-api", "nginx"]

[sources.nginx_access]
type = "file"
include = ["/var/log/nginx/access.log"]

# ── Transformações (transforms) ──────────────────────────
# Parsear JSON de containers que emitem JSON:
[transforms.parse_json]
type = "remap"
inputs = ["docker_logs"]
source = '''
  # Tentar parsear o campo message como JSON:
  parsed, err = parse_json(.message)
  if err == null {
    . = merge(., parsed)
  }
  # Adicionar label para o Loki:
  .service = get_env_var!("HOSTNAME")
'''

# Filtrar logs de health check (ruído):
[transforms.filter_health]
type = "filter"
inputs = ["parse_json"]
condition = '!includes(["GET /health", "GET /metrics"], .message)'

# ── Destinos (sinks) ─────────────────────────────────────
[sinks.loki]
type = "loki"
inputs = ["filter_health"]
endpoint = "${LOKI_URL}"
labels.job = "docker"
labels.container = "{{ container_name }}"
labels.level = "{{ level }}"
encoding.codec = "json"
batch.max_bytes = 1_000_000
batch.timeout_secs = 5

# vector.toml — roteamento por severity:

[transforms.router]
type = "route"
inputs = ["parse_json"]

# Rotas baseadas em condições VRL:
[transforms.router.route]
errors = '.level == "error" || .level == "fatal"'
warnings = '.level == "warn"'
info = '.level == "info" || .level == "debug"'

# Erros → Loki (alta prioridade):
[sinks.loki_errors]
type = "loki"
inputs = ["router.errors"]
endpoint = "${LOKI_URL}"
labels.severity = "error"
labels.container = "{{ container_name }}"
# Flush imediato para erros:
batch.max_bytes = 10_000
batch.timeout_secs = 1

# Todos os logs → S3/MinIO (retenção longa):
[sinks.s3_archive]
type = "aws_s3"
inputs = ["router.errors", "router.warnings", "router.info"]
bucket = "logs-arquivo"
region = "us-east-1"
endpoint = "http://minio:9000"  # MinIO self-hosted
key_prefix = "logs/{{ strftime(to_unix_timestamp(now()), "%Y/%m/%d") }}/"
encoding.codec = "ndjson"
compression = "gzip"
batch.max_bytes = 10_000_000
batch.timeout_secs = 300  # flush a cada 5 min

# Exemplos de transformações VRL comuns:

[transforms.enrich_logs]
type = "remap"
inputs = ["docker_logs"]
source = '''
  # Parsear log do Nginx (formato combined):
  # 177.10.0.1 - - [07/Jun/2026:14:32:10 +0000] "GET /api/users HTTP/1.1" 200 1234
  parsed, err = parse_nginx_log(.message, "combined")
  if err == null {
    .request_method = parsed.method
    .request_path = parsed.path
    .status_code = parsed.status
    .response_bytes = parsed.size
    .remote_addr = parsed.host
  }

  # Extrair tenant de JWT no header:
  if exists(.authorization) {
    jwt, err = parse_jwt(.authorization, algorithm: "HS256")
    if err == null {
      .tenant_id = jwt.sub
    }
  }

  # Mascarar dados sensíveis:
  .message = replace(.message, r'password":"[^"]*"', "password":"***"")

  # Calcular latência em ms a partir de string:
  if exists(.duration) {
    .duration_ms = to_int!(.duration) * 1000
  }

  # Adicionar geo-IP (requer banco MaxMind):
  # geo, err = get_enrichment_table_record("geoip", {"ip": .remote_addr})
  # if err == null { .country = geo.country_code }
'''

# vector.toml — expor métricas do Vector para Prometheus:

[sources.internal_metrics]
type = "internal_metrics"

[sinks.prometheus_exporter]
type = "prometheus_exporter"
inputs = ["internal_metrics"]
address = "0.0.0.0:9598"  # endpoint para scrape do Prometheus

# prometheus.yml — adicionar scrape do Vector:
# - job_name: 'vector'
#   static_configs:
#     - targets: ['localhost:9598']

# Métricas importantes do Vector:
# vector_component_received_events_total  — eventos recebidos por fonte
# vector_component_sent_events_total      — eventos enviados por destino
# vector_component_errors_total           — erros no pipeline
# vector_buffer_events                    — buffer atual (lag)

# Verificar métricas via API:
curl http://localhost:8686/metrics | grep vector_component

# No Grafana — importar dashboard oficial do Vector:
# grafana.com/grafana/dashboards/19649