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.
