Pipeline de Observabilidad — NutriTrack SaaS

# ============================================================
# NutriTrack — Vector Observability Pipeline
# Cliente: NutriTrack SaaS | Env: production
# Generado: 2026-06-18 | CULTIVA IA / IA-Ingenieria-MLOps
# ============================================================

# ---------- SOURCES ------------------------------------------

[sources.api_gateway_logs]
type                 = "file"
include              = ["/var/log/api-gateway/*.log"]
read_from            = "beginning"
data_dir             = "/var/lib/vector/checkpoints"

[sources.api_gateway_logs.multiline]
start_pattern        = '^\\{'      # JSON comienza con {
condition_pattern    = '^\\}'
mode                 = "continue_through"
timeout_ms           = 5000

[sources.nutrition_logs]
type                 = "file"
include              = ["/var/log/nutrition-service/*.log"]
read_from            = "beginning"
data_dir             = "/var/lib/vector/checkpoints"

[sources.billing_logs]
type                 = "file"
include              = ["/var/log/billing-service/*.log"]
read_from            = "beginning"
data_dir             = "/var/lib/vector/checkpoints"

[sources.host_metrics]
type                 = "host_metrics"
collectors           = ["cpu", "memory", "disk", "network"]
scrape_interval_secs = 15

[sources.postgres_logs]
type                 = "syslog"
address              = "0.0.0.0:5140"
mode                 = "tcp"

# ---------- TRANSFORMS ----------------------------------------

# 1. Parsear JSON y añadir metadatos de servicio
[transforms.parse_logs]
type   = "remap"
inputs = ["api_gateway_logs", "nutrition_logs", "billing_logs"]
source = '''
  # Parsear JSON de la línea de log
  . = parse_json!(.message)

  # Determinar servicio desde el path del fichero
  .service = if includes(to_string!(.file ?? ""), "api-gateway") {
    "api-gateway"
  } else if includes(to_string!(.file ?? ""), "nutrition") {
    "nutrition-service"
  } else if includes(to_string!(.file ?? ""), "billing") {
    "billing-service"
  } else {
    .service ?? "unknown"
  }
'''

# 2. Redactar PII/datos personales (LOPD + GDPR)
[transforms.redact_pii]
type   = "remap"
inputs = ["parse_logs"]
source = '''
  # Redactar emails de pacientes (LOPD art.5)
  if exists(.patient_email) {
    .patient_email = redact(.patient_email, filters: ["pattern"],
      redactor: "full",
      patterns: [r'\S+@\S+\.\S+'])
  }
  if exists(.user_email) {
    .user_email = redact(.user_email, filters: ["pattern"],
      redactor: "full",
      patterns: [r'\S+@\S+\.\S+'])
  }
  # Redactar tokens de Stripe en logs de billing
  if exists(.stripe_token) {
    .stripe_token = "[REDACTED]"
  }
  # Redactar IPs reales en modo GDPR
  if exists(.client_ip) {
    parts = split(.client_ip ?? "", ".")
    .client_ip = join!(slice!(parts, 0, 3), ".") + ".0"
  }
'''

# 3. Filtrar ruido: health checks, readiness probes, debug
[transforms.filter_noise]
type      = "filter"
inputs    = ["redact_pii"]
condition = '''
  !includes(["GET /health", "GET /ready", "GET /metrics", "HEAD /"], .path ?? "") &&
  .level != "debug" &&
  !starts_with(.message ?? "", "DeprecationWarning:")
'''

# 4. Sampling: 20% de info, 100% error/warn/fatal
[transforms.sample_logs]
type      = "sample"
inputs    = ["filter_noise"]
rate      = 5            # 1 de cada 5 eventos info (20%)
condition = '.level == "info"'
exclude   = '.level == "error" || .level == "warn" || .level == "fatal"'

# 5. Enriquecer con metadatos de entorno
[transforms.enrich]
type   = "remap"
inputs = ["sample_logs"]
source = '''
  .environment  = "production"
  .app          = "nutritrack"
  .version      = get_env_var("APP_VERSION") ?? "unknown"
  .region       = get_env_var("DEPLOY_REGION") ?? "eu-west-1"
  .log_size_bytes = length(encode_json(.))

  # Normalizar niveles de severidad
  .severity = if .level == "fatal" || .level == "critical" { "error" }
              else if .level == "warning" { "warn" }
              else { .level ?? "info" }

  # Extraer request_id para trazabilidad cross-service
  .trace_id = .request_id ?? .trace_id ?? uuid_v4()
'''

# 6. Agregar métricas de host (ventana de 60s)
[transforms.aggregate_metrics]
type        = "aggregate"
inputs      = ["host_metrics"]
interval_ms = 60000

# ---------- SINKS ----------------------------------------------

# Elasticsearch Cloud — últimos 30 días, búsqueda en tiempo real
[sinks.elasticsearch]
type                = "elasticsearch"
inputs              = ["enrich"]
endpoints           = ["${ES_ENDPOINT}"]
bulk.index          = "nutritrack-logs-%Y-%m-%d"
auth.strategy       = "basic"
auth.user           = "${ES_USER}"
auth.password       = "${ES_PASSWORD}"
compression         = "gzip"
batch.max_bytes     = 10485760   # 10MB
batch.timeout_secs  = 5
request.retry_attempts = 5

[sinks.elasticsearch.buffer]
type     = "disk"          # Buffer en disco para evitar pérdida de datos
max_size = 536870912      # 512MB
when_full = "block"

# S3 — archivo barato a largo plazo (90+ días)
[sinks.s3_archive]
type               = "aws_s3"
inputs             = ["enrich"]
bucket             = "nutritrack-logs-archive"
region             = "eu-west-1"
key_prefix         = "logs/{{ service }}/year=%Y/month=%m/day=%d/"
compression        = "gzip"
encoding.codec     = "json"
batch.max_bytes    = 104857600   # 100MB → ficheros eficientes para Athena
batch.timeout_secs = 300         # Cada 5 minutos

[sinks.s3_archive.auth]
access_key_id     = "${AWS_ACCESS_KEY_ID}"
secret_access_key = "${AWS_SECRET_ACCESS_KEY}"

# Slack #ops-alertas — solo errores críticos en tiempo real
[sinks.slack_errors]
type              = "http"
inputs            = ["enrich"]
uri               = "${SLACK_WEBHOOK_URL}"
method            = "post"
encoding.codec    = "json"
condition         = '.level == "error" || .level == "fatal"'
batch.max_events  = 1             # Alerta inmediata por cada error
request.rate_limit_num = 5        # Máx. 5 mensajes/seg (anti-flood)

# Prometheus — métricas de host para Grafana
[sinks.prometheus]
type    = "prometheus_exporter"
inputs  = ["aggregate_metrics"]
address = "0.0.0.0:9598"

# Habilitar API de healthcheck interno de Vector
[api]
enabled = true
address = "127.0.0.1:8686"

services:
  vector:
    image: timberio/vector:0.37.1-alpine
    restart: unless-stopped
    volumes:
      - ./vector.toml:/etc/vector/vector.toml:ro
      - /var/log/api-gateway:/var/log/api-gateway:ro
      - /var/log/nutrition-service:/var/log/nutrition-service:ro
      - /var/log/billing-service:/var/log/billing-service:ro
      - vector_data:/var/lib/vector     # buffer de disco
    env_file:
      - .env.vector                    # ES_USER, ES_PASSWORD, SLACK_WEBHOOK_URL, AWS_*
    ports:
      - "8686:8686"                    # Vector API / health
      - "9598:9598"                    # Prometheus metrics
      - "5140:5140"                    # Postgres syslog
    command: ["--config", "/etc/vector/vector.toml"]
    healthcheck:
      test: ["CMD", "wget", "-q", "--spider", "http://localhost:8686/health"]
      interval: 30s
      timeout: 5s
      retries: 3
    mem_limit: 128m                   # vs 2GB de Logstash
    cpus: '0.25'

volumes:
  vector_data:

# 1. Validar configuración antes de desplegar
vector validate vector.toml

# 2. Test con VRL REPL (probar transforms sin datos reales)
vector vrl --input test_event.json --program transforms/enrich.vrl

# 3. Arrancar Vector junto al stack de NutriTrack
docker compose -f docker-compose.yml -f docker-compose.vector.yml up -d vector

# 4. Ver logs de Vector en tiempo real
docker compose logs -f vector

# 5. Health check del API interno de Vector
curl -s http://localhost:8686/health | jq .

# 6. Ver estadísticas de throughput en tiempo real
vector top

# 7. Verificar que Prometheus expone las métricas
curl -s http://localhost:9598/metrics | grep host_cpu

# 8. Forzar rotación del buffer de S3 (flush inmediato)
curl -X POST http://localhost:8686/sinks/s3_archive/drop