ELK - Introducción a Logstah III

En el post anterior sobre Logstash, vimos que los dos filtros más utilizados para registros de tipo texto son dissect y grok. Usando syslog_generator, el cual os recuerdo que tenéis disponible aquí por si queréis usarlo, creamos entradas de tipo fijo para el syslog de una máquina donde habíamos instalado filebeat, las enviamos a Logstash y las procesamos con un filtro dissect para obtener los campos que nos interesaban.

En esta entrada usaremos grok sobre el mismo tipo de registros, enviados también desde un sistema con filebeat, para poder realizar la misma tarea.

Primero es importante recordar que grok es perfecto para procesar registros de tipo texto no estructurados, es decir, para aquellos casos en los que se generan registros con una estructura variable de línea a línea. Además, al basarse en expresiones regulares, es posible extender los patrones reconocidos creando patrones para poder usarlos en cualquier pipeline de Logstash que utilice grok.

La sintaxis básica para definir un filtro con grok se define del siguiente modo:

Sintaxis básica para un filtro grok.

Y esta definición, ¿que quiere decir exactamente? Como ya hemos visto, a diferencia de dissect, grok se basa en expresiones regulares ya existentes, identificadas con un nombre que se corresponde con el campo PATTERN, las cuales asignaremos al campo NAME que necesitemos cuando la coincidencia sea satisfactoria. Grok incluye un buen número de patrones ya listos para su uso, cuya lista puede consultarse aquí aunque pueden incluirse nuevos patrones, si es necesario, modificando la configuración de Logstash añadiendo un fichero con la definición de nuevos patrones.

Al definir filtros usando grok, este busca coincidencias desde el principio de cada línea de tecto hasta el final de la misma intentando mapear todo el evento, o hasta no encontrar una coincidencia. En función de los patrones, es posible que grok procese los datos varias veces, por lo que es más lento y requiere más recursos que filtros realizados con dissect.

Para nuestro ejemplo, vamos a seguir con líneas de registro que son perfectas para el uso de dissect, ya que no varían y presentan los mismos campos y delimitadores en todos los casos, pero así podremos comprobar que podemos usar indistintamente cualquiera de los dos tipos de filtro.

Las líneas a procesar presentan la siguiente estructura:

Líneas de registro a procesar y definición de campos.

Al igual que hicimos con el filtro dissect, vamos a empezar con un filtro simple y luego iremos complicándonos un poco más. 

Para empezar, vamos a hacer uso de una herramienta muy útil disponible en Kibana y que nos permite probar nuestros filtros grok, con lo que podemos analizar si el resultado obtenido es o no correcto antes de configurarlo en nuestro pipeline. Podemos acceder al Grok Debugger desde la sección Dev Tools de Kibana:

Grok Debugger disponible en Kibana.

Esta herramienta es de gran ayuda ya que, al tratarse de expresiones regulares, es fácil que tengamos que probar bastantes veces antes de dar con la configuración correcta del filtro. Por tanto, para empezar, vamos a probar a quedarnos solo con 4 campos de la línea message, en concreto con la fecha de syslog, el hostname, el nombre del programa que ha generado la línea y el mensaje generado por syslog_generator. Para esto, el filtro grok a aplicar sobre la línea message será:

Filtro grok inicial.

Como vemos en el filtro, puede que en algunos casos sea necesario definir los espacios que existan en el registro de texto, dados por el patrón SPACE. Teniendo esto en cuenta, el filtro está estableciendo lo siguiente:

• Almacenar en el campo logger_date el texto encontrado que haga match con el patrón SYSLOGTIMESTAMP.
• Almacenar en el campo logger_hostname la siguiente palabra encontrada.
• Almacenar en el campo logger_program los siguientes datos encontrados.
• Almacenar en el campo logger_message todo lo que queda hasta el final de la línea, lo cual especifcamos con el patrón GREEDYDATA.

Usando el grok debugger con este filtro, sobre una línea de ejemplo recibida en Logstash, podemos simular el resultado y comprobar que, al trabajar con expresiones regulares, los resultados que buscamos son un poco más complicados de obtener de lo que esperábamos:

Usando Grok Debugger con el filtro inicial.

Como vemos en la imagen anterior, el campo logger_program se queda vacio, es decir, que la definición %{DATA:logger_program} que hemos hecho para ese campo, no nos devuelve ningún contenido. Si vemos la definición del patrón DATA disponible, vemos que equivale a .*? lo cual establece que debe devolver la menor cantidad posible de caracteres (?) de cualquiera de los caracteres (.*) anteriores. Básicamente, al especificar que se devuelva la menor cantidad posible de caracteres con ?, no está devolviendo ningún valor en dicho campo.

La forma de corregirlo sería usar un filtro como el siguiente, en el cual fijásemos que haga match con cualquier palabra y todos aquellos caracteres que no sean un espacio:

Filtro grok modificado para incluir de forma correcta el cmapo logger_program.

Al probarlo en el depurador de grok de Kibana, vemos que ahora obtenemos los campos que queremos de forma correcta:

Comprobación de que el nuevo filtro es correcto.

La sintaxis empleada es la que podemos usar cuando la expresión regular que necesitamos, no se encuentra en los patrones disponibles incluidos con el plugin de filtro grok de Logstash. Podemos usarla directamente en la definición del filtro o, si queremos usarla de manera habitual, podemos extender la biblioteca de patterns incluyendo las nuestras y asignándoles un nombre para poder usarlas en múltiples pipelines.

Modificamos el fichero de configuración del filtro grok con esta definición de filtro y arrancamos Logstash. La salida que obtenemos en este caso es la siguiente:

Salida de Logstash con el filtro grok correcto.

Podemos ver como los cuatro campos que hemos definido están en la salida, además del resto generados por el plugin de entrada filebeat. Además, el campo message del registro ya no aparece al eliminarlo con la opción remove_field.

Ahora ya solo nos queda procesar el campo logger_message para obtener el resto de campos que necesitamos, para lo cual podríamos aplicar un filtro como el siguiente:

Filtro grok procesando todos los campos de message.

Cuando lo aplicamos, la salida de Logstash ya nos muestra los campos necesarios como podemos ver en la siguiente imagen:

Salida de Logstash con el filtro final grok.

En resumen, está claro que grok nos da mucha más potencia para procesar registros de tipo texto, pero con un coste de dificultad añadido debido al uso de expresiones regulares. En general, el uso de dissect o grok, dependerá de la estructura de los registros de texto que sea necesario procesar y, en algunos casos, será conveniente combinar ambos en el mismo pipeline.

En próximas entradas crearemos más de un pipeline, para ver como trabajan conjuntamente y enviaremos los datos a Elasticsearch para podeer realizar búsquedas sobre los datos recibidos.

Como referencia para el estudio de expresiones regulares, recomiendo usar la página https://regex101.com/ en la cual, además de poder probar expresiones regulares, nos explica cada una de las opciones y operadores que podemos usar para la construcción de expresiones regulares complejas.

ELK - Introducción a Logstah II

Como ya vimos en el post anterior sobre Logstash, este elemento de la pila ELK nos permite recibir datos, procesarlos mediante pipelines y luego enviar dichos datos a otros destinos. Resumiendo el post anterior, los pipelines presentan una estructura que, de forma muy simplificada, podemos ver del siguiente modo:

Definición básica de un pipeline.

Utilizando esta estructura para configurar un pipeline simple, vimos cómo recibir entradas de syslog de un sistema en el cual instalamos filebeat y enviamos dicha salida directamente a STDOUT, sin aplicar ningún filtro a los datos recibidos. Hoy veremos cómo podemos procesar esa información, aplicando filtros a los datos recibidos desde nuestro cliente, para obtener los campos que nos resulten útiles.

Recordando la configuracion que establecimos, al lanzar el generador de entradas de syslog para el tipo FIXED con el pipeline sin filtro, la salida que nos muestra logstash en stdout es como la siguiente:

Salida de Logstash sin filtro a STDOUT.

El objetivo es procesar el campo message, aplicando filtros para poder obtener los valores que realmente necesitamos y que luego podremos enviar a Elasticsearch.

Logstash dispone de múltiples plugins para realizar el filtrado de los datos recibidos, siendo dos de los más usados para registros de tipo texto dissect y grok. Para consultar la lista de plugins de filtro recomiendo visitar la página de Elastic y consultar la documentación disponible sobre Logstash.

Usar dissect o grok dependerá de como sea la estructura de los datos que debemos procesar, más concretamente de si la estructura de los datos está claramente definida. En general:

• Es recomendable usar dissect cuando los registros que recibe logstash siempre contienen los mismos campos separados por delimitadores. Este filtro es muy rápido y es el más sencillo de configurar.
• El filtro grok es ideal para los casos en los que los registros de texto recibidos varían, ya que usa patrones de expresiones regulares para hacer coincidir campos y delimitadores. Este filtro es más complejo de configurar y consume más recursos.

Es importante tener en cuenta que los pipelines de Logstash nos permiten aplicar múltiples filtros a la misma entrada, con lo que podremos pasar los registros recibidos por un filtro para posteriormente aplicarle otro diferente.

Teniendo en cuenta el campo message recibido en cada registro, esta claro que el filtro ideal en este caso es dissect ya que tenemos varios campos separados por delimitadores de longitud fija. Por tanto, podríamos hacer algo como lo siguiente:

Procesado básico de campo message.

El filtro dissect no permitirá mapear cada uno de esos campos y asignarles el nombre que más nos convenga, con lo que podríamos hacer un filtro básico como el siguiente:

Filtro inicial empleando dissect.

Si analizamos el filtro anterior vemos los siguientes puntos importantes:

• Un campo del filtro es la clave que hay entre los caracteres %{}, con lo que en este caso establecemos 6 campos. Cada uno de estos campos se corresponde con los 6 campos que hemos identificado sobre el mensaje recibido. Asignamos un nombre diferente a cada uno de ellos, según nuestras necesidades.
• El delimitador es cualquier carácter entre los caracteres } y %{, siendo en este caso espacios.
• Como el número de espacios después del campo timestamp_month es variable, añadimos el sufijo -> tras el nombre del campo.
• Además, al emplear la opción remove_field, dissect eliminrá el campo message siempre que se produzca una coincidencia del filtro, con lo que eliminaremos el campo que procesamos de la salida dada.

Al aplicar esta configuración al pipeline de Logstash, la salida que obtenemos ahora es la siguiente:

Salida de Logstash con filtro dissect básico.

Al aplicar el filtro dissect, aparecen los nuevos campos que hemos definido en el filtro, además de todos los campos que genera el plugin de entrada de filebeat con la información del cliente remoto y ha desaparecido el campo message que hemos procesado.

Ya hemos avanzado en el procesado de los registros de syslog que recibimos, pero suponiendo que la información que necesitamos es realmente la que está en el campo logger_message, podemos cambiar el filtro, asignar unos nombres más claros a los campos que nos interesan y trocear el campo logger message con el siguiente filtro:

Ampliamos el filtro dissect al campo logger_message.

Con lo que la nueva salida de logstash es la siguiente:

Salida de Logstash tras procesar el campo logger_message.

Como vemos en la salida anterior, ahora tenemos un conjunto de campos identificados con el prefijo logger_ así como los campos identificados con el prefijo timestamp_. Suponiendo que los campos con el prefijo logger_ de fecha y hora son los que necesitamos además de unificarlos todos en un solo campo y eliminando los campos con el prefijo timestamp_, el filtro final que hacemos con dissect sería el siguiente:

Eliminamos campos y unificamos los campos de fecha en uno solo.

Al aplicar el filtro anterior, la salida de Logstash es la siguiente:

Salida de Logstash eliminando campos timestamp_ y unificando campos de fecha.

Cómo podemos ver en la salida anterior, ya no aparece ningún campo timestamp_ y todos los campos de fecha, con el prefijo logger_, están unificados en el campo logger_timestamp. Para conseguir esto, solo debemos tener en cuenta los siguientes puntos sobre el filtro que hemos definido:

• Para eliminar totalmente un campo de la salida del filtro dissect, solo tenemos que prefijar el nombre del campo con el carácter ?. Como vemos en el filtro anterior, he incluido el carácter ? a todos los campos timestamp_ para eliminarlos.
• Si queremos concatenar campos, basta con repetir el nombre del campo añadiendo el carácter + delante del nombre. Así, en el filtro aplicado, podemos ver como todos los campos que incluyen información de fecha tienen el nombre logger_timestamp correspondiente al primer campo de fecha. El resto de campos que queremos concatenar, repiten el nombre del campo inicial incluyendo el caracter + delante del nombre, indicando así con que campo deben concatenarse.

En la próxima entrada sobre Logstash analizaremos como podemos realizar el mismo filtrado usando grok el cual, como veremos, es un poco más complejo y a la vez más potente.