Logstash, un componente esencial en la pila ELK (Elasticsearch, Logstash, Kibana), se distingue por sus robustas capacidades de procesamiento de datos, en gran medida gracias a su sistema de plugins de filtro. Estos filtros permiten transformar, enriquecer y estructurar datos no procesados en formatos útiles para análisis y almacenamiento.

Extracción con Grok y Expresiones Regulares

El plugin de filtro Grok es una herramienta fundamental en Logstash para parsear datos no estructurados y convertirlos en información estructurada y fácilmente indexable. Este plugin utiliza expresiones regulares para extraer campos específicos de cadenas de texto.

Sintaxis Básica de Grok

La sintaxis (?<nombre_campo>patrón_regex) permite asignar un nombre al valor capturado por el patrón de expresión regular.

Ejemplo 1: Extracción de un Patrón Numérico

Imaginemos que necesitamos extraer un valor numérico con decimales de una entrada simple en la consola.

input { stdin { } }
filter {
  grok {
    match => {
      "message" => "(?<valor_decimal>\\d+\\.\\d+)\\s+"
    }
  }
}
output { stdout { codec => rubydebug } }

Extracción de Múltiples Campos Personalizados

Ampliando el ejemplo anterior, podemos extraer múltiples componentes de una cadena de texto y asignarles nombres significativos a cada uno de ellos.

Ejemplo 2: Parsing de una Frace en Componentes

Para una entrada como "4.19 es un día de suerte", podemos extraer cada palabra como un campo distinto.

input { stdin { } }
filter {
  grok {
    match => {
      "message" => "(?<dia_mes>\\d+\\.\\d+)\\s+(?<verbo>\\w+)\\s+(?<adjetivo>\\w+)\\s+(?<sustantivo>\\w+)"
    }
  }
}
output { stdout { codec => rubydebug } }

Uso de Patrones Prediseñados en Grok

Logstash incluye una extensa biblioetca de patrones Grok predefinidos que simplifican el proceso de parsing para formatos comunes como direcciones IP, números, palabras, fechas, etc. Estos patrones se pueden referenciar directamente en la configuración.

Ejemplo 3: Reescritura del Ejemplo Anterior con Patrones Prediseñados

input { stdin { } }
filter {
  grok {
    match => {
      "message" => "%{NUMBER:numero_evento:float} %{WORD:primer_termino} %{WORD:segundo_termino} %{WORD:tercer_termino}"
    }
  }
}
output { stdout { codec => rubydebug } }

Estructuración de Datos No Estructurados con Grok: Logs de Nginx

Los logs generados por servidores web como Nginx suelen estar en un formato de texto plano que, aunque legible para humanos, es ineficiente para el análisis automatizado. Logstash, con su plugin Grok, permite transformar estos logs no estructurados en datos estructurados y fáciles de consultar en tiempo real, evitando la necesidad de herramientas de procesamiento por lotes adicionales.

Ejemplo de Log de Nginx:

192.168.1.100 - - [15/Oct/2023:09:30:15 +0200] "GET /api/status HTTP/1.1" 200 120 "http://www.ejemplo.com/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36" "-"

Configuración de Grok para Logs de Nginx:

input { stdin { } }
filter {
  grok {
    match => {
      "message" => "%{IPORHOST:ip_cliente} - - \\[\\%{HTTPDATE:marca_temporal}\\] \\"(?:%{WORD:metodo_http} %{NOTSPACE:ruta_recurso}(?: HTTP/%{NUMBER:version_http})?|%{DATA:peticion_cruda})\\" %{NUMBER:codigo_estado} %{NUMBER:body_bytes_sent} %{QS:referencia_http} %{QS:agente_usuario} %{NOTSPACE:reenviado_por}"
    }
  }
}
output { stdout { codec => rubydebug } }

Nota: La expresión regular de Grok debe adaptarse específicamente al formato de log de Nginx que se esté utilizando.

Geolocalización con el Plugin GeoIP

Una vez que los logs están estructurados y hemos extraído la dirección IP del cliente, podemos enriquecer estos datos con información geográfica utilizando el plugin logstash-filter-geoip. Este filtro determina la ubicación física (país, región, ciudad, coordenadas) basándose en una dirección IP, siendo invaluable para análisis de seguridad, marketing o rendimiento.

Ejemplo de Configuración GeoIP Básica:

input { stdin { } }
filter {
  grok {
    match => {
      "message" => "%{IPORHOST:ip_origen} - - \\[\\%{HTTPDATE:fecha_acceso}\\] \\"(?:%{WORD:metodo} %{NOTSPACE:ruta}(?: HTTP/%{NUMBER:protocolo_ver})?|%{DATA:peticion_completa})\\" %{NUMBER:estado_respuesta} %{NUMBER:tamano_bytes} %{QS:referencia_url} %{QS:cadena_agente} %{NOTSPACE:proxy_info}"
    }
  }
  geoip {
    source => "ip_origen"  # Campo de la IP a analizar
    target => "geo_info"   # Nombre del campo para guardar los resultados de geolocalización
  }
}
output { stdout { codec => rubydebug } }

Uso de Bases de Datos GeoIP Personalizadas y Selección de Campos

Para mayor precisión o para utilizar versiones actualizadas de las bases de datos de geolocalización, se puede especificar una base de datos GeoLite2-City.mmdb personalizada. Además, es posible seleccionar solo los campos geográficos relevantes y reestructurar las coordenadas para sistemas de mapas específicos.

Ejemplo de Configuración GeoIP Avanzada:

input { stdin { } }
filter {
  grok {
    match => {
      "message" => "%{IPORHOST:host_ip} - - \\[\\%{HTTPDATE:fecha_hora_log}\\] \\"(?:%{WORD:verbo_http} %{NOTSPACE:url_solicitada}(?: HTTP/%{NUMBER:version_proto})?|%{DATA:solicitud_bruta})\\" %{NUMBER:estado_respuesta_num} %{NUMBER:bytes_transferidos} %{QS:url_referencia} %{QS:info_navegador} %{NOTSPACE:ip_proxy}"
    }
  }
  geoip {
    source    => "host_ip"
    database  => "/ruta/a/tu/GeoLite2-City.mmdb" # Ruta a tu base de datos GeoLite2
    target    => "ubicacion_geo"
    add_field => { "[ubicacion_geo][coordenadas]" => "%{[ubicacion_geo][longitude]}" }
    add_field => { "[ubicacion_geo][coordenadas]" => "%{[ubicacion_geo][latitude]}" }
    fields    => ["country_name", "region_name", "city_name", "latitude", "longitude", "timezone"]
  }
}
output { stdout { codec => rubydebug } }

División de Datos Clave-Valor con el Plugin KV

Los datos a menudo se presentan en formatos clave-valor dentro de una cadena, como los parámetros de consulta de una URL. El plugin kv de Logstash es ideal para parsear estas cadenas y extraer cada par clave-valor en campos separados.

Ejemplo: Parsing de Parámetros URL

Para una URL como https://www.ejemplo.com/pagina?param1=valorA&param2=valorB&param3=valorC, podemos extraer param1, param2 y param3 como campos individuales.

input { stdin { } }
filter {
  kv {
    prefix      => "param_"     # Prefijo para los nombres de campo extraídos
    source      => "message"    # Campo de origen que contiene la cadena clave-valor
    field_split => "&"        # Carácter que separa los pares clave-valor (aquí el '&')
    value_split => "="        # Carácter que separa la clave del valor
  }
}
output { stdout { codec => rubydebug } }