Vector: Recolección y Transformación de Logs

Vector es una herramienta de observabilidad de alto rendimiento escrita en Rust que permite recolectar, transformar y enrutar logs, métricas y trazas con una latencia mínima. A diferencia de Logstash o Fluentd, Vector ofrece un consumo de recursos notablemente menor y un lenguaje de transformación propio (VRL) que facilita el procesamiento complejo de datos. Esta guía cubre la instalación en Linux, configuración de pipelines y la integración con stacks de monitoreo como Loki, Elasticsearch y Datadog.

Requisitos Previos

Ubuntu 20.04/22.04, Debian 11+ o CentOS 8+/Rocky Linux 8+
512 MB de RAM mínimo (Vector es muy eficiente)
Acceso a las fuentes de logs (archivos, Docker, syslog, etc.)
Destino donde enviar los logs (Loki, Elasticsearch, S3, etc.)

Instalación de Vector

Instalación con el script oficial

# Método más sencillo: script de instalación oficial
curl --proto '=https' --tlsv1.2 -sSf https://sh.vector.dev | bash

# Verificar instalación
vector --version

Instalación con apt (Ubuntu/Debian)

# Añadir repositorio oficial de Vector
curl -1sLf 'https://repositories.timber.io/public/vector/cfg/setup/bash.deb.sh' | sudo bash

# Instalar Vector
sudo apt install -y vector

# Habilitar e iniciar el servicio
sudo systemctl enable vector --now

# Verificar estado
sudo systemctl status vector

Instalación con rpm (CentOS/Rocky)

# Añadir repositorio
curl -1sLf 'https://repositories.timber.io/public/vector/cfg/setup/bash.rpm.sh' | sudo bash

# Instalar
sudo dnf install -y vector

# Habilitar e iniciar
sudo systemctl enable vector --now

Instalación con Docker

# Ejecutar Vector como contenedor
docker run \
  -d \
  --name vector \
  -v /etc/vector/vector.yaml:/etc/vector/vector.yaml:ro \
  -v /var/log:/var/log:ro \
  timberio/vector:latest-alpine

Conceptos Fundamentales

Vector organiza el procesamiento de datos en una topología con tres componentes:

Sources: fuentes de datos (archivos de log, Docker, syslog, HTTP, Kafka...)
Transforms: transformaciones aplicadas a los datos (parsing, filtrado, enriquecimiento...)
Sinks: destinos donde se envían los datos procesados (Loki, Elasticsearch, S3, stdout...)

El archivo de configuración principal se encuentra en /etc/vector/vector.yaml.

Configuración de Fuentes (Sources)

# /etc/vector/vector.yaml

# Leer archivos de log del sistema
sources:
  archivos_nginx:
    type: file
    include:
      - /var/log/nginx/access.log
      - /var/log/nginx/error.log
    read_from: beginning   # O "end" para solo logs nuevos

  # Leer logs de contenedores Docker
  docker_logs:
    type: docker_logs
    include_containers:
      - mi-app
      - mi-api

  # Leer desde syslog
  syslog_udp:
    type: syslog
    mode: udp
    address: "0.0.0.0:514"

  # Recibir datos via HTTP (webhook)
  http_webhook:
    type: http_server
    address: "0.0.0.0:8080"
    encoding:
      codec: json

  # Leer desde Kafka
  kafka_input:
    type: kafka
    bootstrap_servers: "kafka:9092"
    group_id: vector-consumer
    topics:
      - logs-aplicacion
    decoding:
      codec: json

Transformaciones con VRL

VRL (Vector Remap Language) es el lenguaje de transformación propio de Vector, diseñado para ser seguro y performante:

transforms:
  # Parsear logs de Nginx con expresión regular
  parsear_nginx:
    type: remap
    inputs:
      - archivos_nginx
    source: |
      # Extraer campos del log de acceso de Nginx
      parsed, err = parse_regex(.message, r'(?P<ip>\S+) \S+ \S+ \[(?P<timestamp>[^\]]+)\] "(?P<metodo>\S+) (?P<ruta>\S+) \S+" (?P<status>\d+) (?P<bytes>\d+)')
      if err != null {
        log("Error parseando log: " + string!(err), level: "warn")
        abort
      }
      . = merge(., parsed)
      .status = to_int!(.status)
      .bytes = to_int!(.bytes)
      # Añadir campo de nivel según el código HTTP
      if .status >= 500 {
        .nivel = "error"
      } else if .status >= 400 {
        .nivel = "warning"
      } else {
        .nivel = "info"
      }

  # Filtrar logs: solo errores HTTP 5xx
  solo_errores:
    type: filter
    inputs:
      - parsear_nginx
    condition: '.status >= 500'

  # Enriquecer con geolocalización IP
  enriquecer_geo:
    type: remap
    inputs:
      - parsear_nginx
    source: |
      # Resolver geolocalización de la IP del cliente
      .geo = get_enrichment_table_record!("geoip", {"ip": .ip})

  # Deduplicar eventos repetidos
  deduplicar:
    type: dedupe
    inputs:
      - parsear_nginx
    fields:
      match:
        - ip
        - ruta
        - status
      cache:
        max_events: 5000

  # Agregar métricas desde logs
  metricas_nginx:
    type: log_to_metric
    inputs:
      - parsear_nginx
    metrics:
      - type: counter
        field: status
        name: nginx_peticiones_total
        tags:
          status: "{{status}}"
          metodo: "{{metodo}}"

Destinos (Sinks)

sinks:
  # Enviar a Loki (Grafana)
  loki:
    type: loki
    inputs:
      - parsear_nginx
    endpoint: http://loki:3100
    encoding:
      codec: json
    labels:
      job: nginx
      host: "{{ host }}"
      nivel: "{{ nivel }}"

  # Enviar a Elasticsearch / OpenSearch
  elasticsearch:
    type: elasticsearch
    inputs:
      - parsear_nginx
    endpoints:
      - http://elasticsearch:9200
    index: "logs-nginx-%Y-%m-%d"
    auth:
      strategy: basic
      user: elastic
      password: contraseña_elastic

  # Enviar a S3 para archivado a largo plazo
  s3_archivo:
    type: aws_s3
    inputs:
      - parsear_nginx
    bucket: mi-bucket-logs
    region: eu-west-1
    key_prefix: "logs/nginx/%Y/%m/%d/"
    encoding:
      codec: json
    compression: gzip
    batch:
      max_bytes: 10000000   # Archivos de ~10 MB
      timeout_secs: 300     # O cada 5 minutos

  # Enviar métricas a Prometheus (via remote_write)
  prometheus:
    type: prometheus_remote_write
    inputs:
      - metricas_nginx
    endpoint: http://prometheus:9090/api/v1/write

  # Enviar a Datadog
  datadog:
    type: datadog_logs
    inputs:
      - solo_errores
    default_api_key: "${DATADOG_API_KEY}"
    site: datadoghq.eu

  # Salida por consola (útil para depuración)
  consola:
    type: console
    inputs:
      - parsear_nginx
    encoding:
      codec: json

Ejemplo de Pipeline Completo

Aquí un pipeline funcional completo para recolectar logs de aplicaciones Docker y enviarlos a Loki:

# /etc/vector/vector.yaml - Pipeline completo para Docker + Loki

sources:
  docker:
    type: docker_logs
    include_labels:
      vector.enable: "true"

transforms:
  parsear_json:
    type: remap
    inputs:
      - docker
    source: |
      # Intentar parsear el mensaje como JSON estructurado
      structured, err = parse_json(.message)
      if err == null {
        . = merge(., structured)
      }
      # Normalizar el campo de nivel de log
      .nivel = downcase(string!(.level ?? .severity ?? .lvl ?? "info"))
      # Eliminar campos internos de Docker que no necesitamos
      del(.stream)
      del(.source_type)

  filtrar_health:
    type: filter
    inputs:
      - parsear_json
    condition: '!includes(["GET /health", "GET /ping"], .message)'

sinks:
  loki:
    type: loki
    inputs:
      - filtrar_health
    endpoint: http://loki:3100
    encoding:
      codec: json
    labels:
      job: "{{ container_name }}"
      nivel: "{{ nivel }}"
      host: "{{ host }}"
    batch:
      max_bytes: 1048576
      timeout_secs: 5

# Validar la configuración antes de aplicarla
vector validate /etc/vector/vector.yaml

# Probar la configuración en modo dry-run
vector tap --config /etc/vector/vector.yaml

# Recargar configuración sin reiniciar
sudo systemctl reload vector

# Ver métricas internas de Vector
vector top

Solución de Problemas

Vector no lee archivos de log:

# Verificar permisos del usuario vector sobre los archivos
ls -la /var/log/nginx/
sudo usermod -aG adm vector  # Añadir vector al grupo que lee logs

# Reiniciar el servicio
sudo systemctl restart vector

Los logs no llegan al destino:

# Revisar logs de Vector
sudo journalctl -u vector -f

# Activar modo debug temporalmente
VECTOR_LOG=debug vector --config /etc/vector/vector.yaml

# Usar sink de consola para depurar el pipeline
# Añadir temporalmente:
# sinks:
#   debug:
#     type: console
#     inputs: [nombre_transform]
#     encoding:
#       codec: json

Error de conexión a Loki/Elasticsearch:

# Verificar conectividad de red
curl -v http://loki:3100/ready
curl -v http://elasticsearch:9200/_cluster/health

# Revisar que las URLs en la configuración sean correctas
vector validate --config /etc/vector/vector.yaml

Consumo de memoria elevado:

# Limitar el buffer de Vector en la configuración del sink
sinks:
  loki:
    # ... configuración ...
    buffer:
      type: memory
      max_events: 500
      when_full: drop_newest

Conclusión

Vector es una herramienta excepcional para construir pipelines de observabilidad eficientes gracias a su bajo consumo de recursos, su velocidad y la potencia de VRL para transformaciones complejas. Su arquitectura declarativa permite definir topologías completas en un solo archivo YAML, facilitando el mantenimiento y la gestión como código. Integrado con herramientas como Loki, Elasticsearch o Datadog, Vector se convierte en la pieza central de cualquier stack de observabilidad moderno.

Vector Log Collection and Transformation

On this page