El stack ELK (ElasticSearch-Logstash-Kibana) es uno de los estándares para el tratamiento y almacenamiento centralizado de logs y visualización de eventos. En este post, veremos una forma fiable de creación y mantenimiento de filtros de Logstash con logstash-filter-verifier.

Introducción

El funcionamiento básico del stack ELK es el siguiente:

• Beats: Se encargan de recoger información que poder enviar para indexar. El beat más usado y en el que nos centraremos en este post es Filebeat. Filebeat se encarga de recoger esa información desde ficheros (logs).
• Logstash: Se encarga de recibir logs de diversas fuentes (generalmente, los beats), identificar campos dentro de las lineas de log e indexarlo dentro de ElasticSearch como un documento.
• ElasticSearch: base de datos NoSQL basado en Lucene donde se indexarán los documentos de los logs.
• Kibana: visualización de información almacenada en ElasticSearch.

Generalmente, la curva de aprendizaje de este Stack es más pronunciada en el componente de Logstash. Esto se debe, probablemente, a que Logstash ejecuta la tarea más importante dentro del stack: extraer campos del log e indexarlos en ElasticSearch. Esta indexación es lo que de verdad aporta valor en este stack frente a, por ejemplo, un rsyslog. Cuando la indexación falla, lo más probable es que muchos de los dashboards de Kibana dejen de mostrar información al no encontrar el campo correspondiente en los documentos de ElasticSearch, las búsquedas por campo dejarán de ser fiables por el sesgo de datos, etc.

La tarea de depuración de los filtros de Logstash puede llegar a ser bastante complicada y, conforme la cantidad de filtros aumenta, el mantenimiento de estos filtros en el largo plazo se hace cada vez más complicado (sobre todo si compartes filtros entre distintas aplicaciones).

TDD + logstash-filter-verifier para creación de filtros

Con el objetivo de simplificar tanto la programación de filtros como de mantenerlos en el largo plazo, usaremos logstash-filter-verifier. Esta utilidad nos da la posibilidad de realizar tests unitarios sobre nuestras configuraciones de Logstash. Además, si trabajamos con TDD (Test-Driven-Development) para la creación de los filtros, no solo conseguimos que la depuración y creación de filtros sea más fácil sino que, además, detectamos fallos de configuración antes de añadir el cambio.

A modo de ejemplo, supongamos que queremos crear un filtro que parsee esta líneas de log:

Oct 6 20:55:29 myhost myprogram[31993]: This is a test message

Para hacer este parseo de campos necesitaríamos hacer lo siguiente:

1. Creamos el siguiente filtro vacío:

# conf.d/03-0-sample-filter.conf
filter {
}

2. Definimos el test-case con los campos que queremos indexar en formato json como lo espera recibir logstash-filter-verifier:

# tests/03-0-sample-filter-testcase.json
{
  "testcases": [
    {
      "input": [
        "Oct 6 20:55:29 myhost myprogram[31993]: This is a test message"
      ],
      "expected": [
        {
          "@timestamp": "2021-10-06T20:55:29.000Z",
          "host": "myhost",
          "message": "This is a test message",
          "pid": 31993,
          "program": "myprogram"
        }
      ]
    }
  ]
}

3. Descargamos logstash y logstash-filter-verifier si no los tenemos ya instalados:

# Logstash binary
$ cd /opt
$ sudo wget https://artifacts.elastic.co/downloads/logstash/logstash-7.12.0-linux-x86_64.tar.gz
$ sudo tar -xvzf logstash-7.12.0-linux-x86_64.tar.gz
$ ln -s /opt/logstash-7.12.0/ /opt/logstash

# logstash-filter-verifier
$ cd /usr/local/bin
$ sudo wget https://github.com/magnusbaeck/logstash-filter-verifier/releases/download/1.6.2/logstash-filter-verifier_1.6.2_linux_amd64.tar.gz
$ sudo tar -xvzf logstash-filter-verifier_1.6.2_linux_amd64.tar.gz

4. Ejecutamos el test case:

$ logstash-filter-verifier tests/03-0-sample-testcase.json conf.d/
Running tests in 03-0-sample-testcase.json...
☐ Comparing message 1 of 1 from 03-0-sample-testcase.json:
--- /tmp/620401130/03-0-sample-testcase.json/1/expected 2021-04-20 20:09:35.514280632 +0200
+++ /tmp/620401130/03-0-sample-testcase.json/1/actual   2021-04-20 20:09:35.514280632 +0200
@@ -1,7 +1,5 @@
 {
-  "@timestamp": "2021-10-06T20:55:29.000Z",
-  "host": "myhost",
-  "message": "This is a test message",
-  "pid": 31993,
-  "program": "myprogram"
+  "@timestamp": "2021-04-20T20:09:34.253Z",
+  "host": "...",
+  "message": "Oct 6 20:55:29 myhost myprogram[31993]: This is a test message"
 }


Summary: ☐ All tests: 0/1
         ☐ 03-0-sample-testcase.json: 0/1

5. Obviamente, como no implementamos el filtro, el test unitario falló (como siempre que se usa TDD). Pasemos entonces a la implementación de este filtro y veamos cómo podemos obtener los campos del log:

# conf.d/03-0-sample-filter.conf
filter {
  grok {
    match => {
      "message" => "%{SYSLOGTIMESTAMP:[@metadata][timestamp]} %{SYSLOGHOST:host} %{SYSLOGPROG:pid:int}: %{GREEDYDATA:message}"
    }
    overwrite => ["message", "host", "pid"]
  }
  date {
    match => [
      "[@metadata][timestamp]", "MMM d HH:mm:ss", "MMM dd HH:mm:ss", "MMM dd yyyy HH:mm:ss", "MMM d yyyy HH:mm:ss", "dd/MMM/yyyy:HH:mm:ss Z", "ISO8601"
    ]
    remove_field => "timestamp"
  }
}

6. Con este filtro conseguimos sacar los valores de los campos host, pid y message. Además, sobreescribimos el timestamp para usar el del propio log (en vez de usar el timestamp de recepción en logstash). Volvemos a ejecutar logstash-pipeline-verifier para ver si con este filtro pasamos el test:

$ logstash-filter-verifier tests/03-0-sample-testcase.json conf.d/
Running tests in 03-0-sample-testcase.json...
☐ Comparing message 1 of 1 from 03-0-sample-testcase.json:
--- /tmp/532357049/03-0-sample-testcase.json/1/expected 2021-04-20 20:18:59.408976861 +0200
+++ /tmp/532357049/03-0-sample-testcase.json/1/actual   2021-04-20 20:18:59.408976861 +0200
@@ -2,6 +2,6 @@
   "@timestamp": "2021-10-06T20:55:29.000Z",
   "host": "myhost",
   "message": "This is a test message",
-  "pid": 31993,
+  "pid": "31993",
   "program": "myprogram"
 }


Summary: ☐ All tests: 0/1
         ☐ 03-0-sample-testcase.json: 0/1

7. En este caso, seguimos sin pasar el test porque el pid que indicamos en nuestro test-case esperaba recibir un número en vez de una cadena de caracteres. En el caso del pid, no es demasiado importante forzar que sea un entero. Sin embargo, si fuese una métrica de temporal, usar una cadena implicaría que no se podrían aplicar operaciones matemáticas directamente, algo muy perjudicial para ciertas visualizaciones de datos. Ahora sí, hacemos el cambio oportuno y pasamos el test:

$ git diff
diff --git a/conf.d/03-0-sample-filter.conf b/conf.d/03-0-sample-filter.conf
index b4eb69f..4869937 100644
--- a/logstash/conf.d/03-0-sample-filter.conf
+++ b/logstash/conf.d/03-0-sample-filter.conf
@@ -13,5 +13,8 @@ filter {
       ]
       remove_field => "timestamp"
     }
+    mutate {
+      convert => { "pid" => "integer" }
+    }
   }
 }

$ logstash-filter-verifier tests/03-0-sample-testcase.json conf.d/
Running tests in 03-0-sample-testcase.json...
☑ Comparing message 1 of 1 from 03-0-sample-testcase.json

Summary: ☑ All tests: 1/1
         ☑ 03-0-sample-testcase.json: 1/1

Automatización de tests unitarios con GitHub Actions

Para realizar todos los tests de logstash-filter-verifier cada vez que se sube un cambio al repositorio de GitHub, podemos usar un workflow de GitHub Actions como este:

# .github/workflows/logstash-filter-verifier.yml
name: Logstash Filters Unit Tests
on:
  workflow_dispatch:
  push:
jobs:
  unittest:
    runs-on: ubuntu-latest
    container:
      image: logstash:7.11.2
      options: --user root
    env:
      LOGSTASH_FILTER_VERIFIER_VERSION: 1.6.2
    steps:
    - name: Checkout code
      uses: actions/checkout@v2

    - name: Download logstash-filter-verifier
      run: |
        curl -LO https://github.com/magnusbaeck/logstash-filter-verifier/releases/download/${{ env.LOGSTASH_FILTER_VERIFIER_VERSION }}/logstash-filter-verifier_${{ env.LOGSTASH_FILTER_VERIFIER_VERSION }}_linux_amd64.tar.gz
        tar -xvzf logstash-filter-verifier*

    - name: Run tests
      run: |
        ./logstash-filter-verifier tests/ conf.d/

Conclusiones

El mantenimiento de los filtros de logstash puede llegar a ser muy dificil de mantener en el largo plazo. Tener una herramienta que realice directamente los tests unitarios nos permite tener la seguridad de todo momento que ante nuevos cambios no estamos dañando el procesado de otros logs.

Además, el hecho de extraer muestras de logs y definir los test-cases antes de implementarlo, permite no solo que la robustez de los filtros sea mucho superior, sino también que la interacción con otros filtros sea directa. Por poner un ejemplo, es muy común tener un filtro genérico que parsee todos los logs de un servidor Apache/Httpd. La utilización de campos generados desde esos filtros es directa si seguimos la metodología TDD (directamente veremos los campos de los que queremos leer).

En resumen, el TDD y los tests unitarios permiten que la forma de trabajar con Logstash sea mucho más eficiente y robusta, y su curva de aprendizaje sea mucho menor. ¿Quieres dejar de depurar tags _grok_parse_failure? Usa logstash-filter-verifier.

Álvaro Torres Cogollo.

¿Quieres contactar conmigo? Te dejo mis redes sociales a continuación.